Distributed query execution — extended reference

This reference preserves the earlier detailed chapter and its citations. Start with the guided explanation, then use these sections for additional mechanisms and qualifications. References to the original lesson sequence are historical.

Part 6: Scaling execution and state delivery

Prerequisites: Lesson 30 and the preceding foundations.

Learning outcomes

Follow distributed scans, joins, exchanges, and partial aggregation through data movement.
Calculate broadcast and repartition costs and handle weighted aggregate state correctly.
Explain partition pruning, skew, stragglers, and the difference between transactional and analytical priorities.

The network can carry far more intermediate data than answers

A global report joins a large fact table with a small dimension and returns ten grouped totals. The final answer fits in a few hundred bytes. A poor execution plan moves hundreds of gigabytes between workers before producing it. Distributed query design is therefore primarily about where data is processed and how much must move, not how small the result looks.

Parallelism can reduce elapsed time while increasing total resource consumption. A query that completes faster by occupying every node may harm transactional traffic. The engine needs partition-aware planning, exchange operators, memory budgets, skew handling, and cancellation across all participants.

Partition pruning and colocated work

If a predicate identifies specific partitions, the engine can avoid contacting irrelevant owners. A tenant-local query benefits when its data and indexes align with tenant ownership. A time-range query benefits from range metadata that excludes older or newer partitions.

Pruning depends on query semantics and metadata. An expression that obscures the partition key may prevent a simple range deduction. Stale placement information must be handled safely through routing or retries. A correct query should not silently omit a partition because a cached map is old.

Colocated joins keep matching data on the same workers under compatible partitioning. This can reduce network exchanges, but maintaining colocation constrains placement and rebalancing. A skewed tenant can overload one worker even when the rest of the cluster is idle. Locality and balance are competing physical properties.

Broadcast joins

A broadcast join sends a smaller input to workers processing the larger input. Each worker builds or accesses a local representation and joins its own fact partition. This avoids repartitioning the large input when the dimension is truly small enough.

For an illustrative 20 MB dimension broadcast to fifty workers, payload transfer is about 1 GB before protocol overhead. Each worker also needs memory for its local representation, potentially larger than raw input because of hash metadata. Broadcasting several such dimensions multiplies both network and memory demands.

A dimension that is small in row count can still be wide. A skewed or poorly estimated filter can make a planned broadcast unexpectedly large. The engine needs thresholds and a safe response when runtime size exceeds expectations. Treating a broadcast plan as unconditionally cheap can exhaust a cluster.

Repartition joins

A repartition join hashes or ranges both inputs by the join key so matches meet at the same worker. Exchange operators move rows between stages. This can distribute work across the cluster while creating a large shuffle.

left partitions  -> hash(join_key) -> destination workers
right partitions -> hash(join_key) -> destination workers
                                      |
                                  local joins

The diagram shows a logical exchange. Implementations may pipeline, buffer, or spill the transfer. Every sender and receiver needs bounded queues. A fast producer must not force an unbounded receiver allocation, and a failed worker must trigger coherent cleanup or retry.

A hot join key can send a disproportionate fraction of rows to one destination. Rehashing the same equality key does not remove that concentration. Skew-aware strategies can isolate heavy keys, replicate selected sides, or split work with additional bookkeeping. The output multiplicity still imposes a lower bound when many rows genuinely match.

Partial aggregation

Workers can aggregate locally before sending intermediate state. For a sum grouped by country, each worker sends one partial sum per country instead of every payment row. The final stage combines partials using a valid aggregate algebra.

An average requires sum and count. Suppose one worker sees two values averaging 100 and another sees 100 values averaging 1. The unweighted average of worker averages is 50.5, while the correct combined average is (200+100)/102, approximately 2.941. The example shows why reducing traffic must preserve weighting and null semantics.

Distinct counts can require large state or an explicitly approximate sketch. A mergeable sketch has an error contract and often a randomized component. Calling the result a count without describing approximation would change the query’s meaning. Exactness is part of the planning constraint.

Exchanges, order, and memory

An exchange can preserve, merge, or destroy ordering depending on its type. A final ORDER BY may require a global merge or sort even when each worker’s local input is ordered. A limit can sometimes reduce traffic through distributed top-k strategies, but correctness requires accounting for which local candidates can affect the global result.

Memory is consumed at both ends of an exchange. Serialization buffers, receiver queues, hash tables, and spill files can overlap. Per-worker limits should be composed into a query and cluster budget. A plan with fifty workers can multiply a seemingly modest local reservation into substantial total demand.

Compression can reduce shuffle bytes while consuming CPU. It may be beneficial across a constrained network and less useful for tiny messages between nearby workers. The cost model should include the actual bottleneck and the time to first useful output.

Stragglers and skew

The slowest required worker can determine query completion. A straggler may have more data, colder caches, slower storage, a busy neighbor, or a pause. Treating every slow worker as a hardware failure can trigger unnecessary retries and extra load.

Speculative execution can help for safely repeatable work, but it duplicates resource use and needs result deduplication. A task that performs mutations requires a stronger protocol than an immutable scan. Cancellation should stop losing copies where possible, while recognizing that some work may already have completed.

Dynamic scheduling can assign smaller chunks to workers as they become available. This improves balance when chunks are independent and metadata overhead is manageable. Too-small chunks increase scheduling and setup cost. The useful granularity follows the work’s variability and resource profile.

Transactions versus analytical priorities

A long analytical scan may favor large batches and high parallelism. A transactional lookup favors bounded waiting and fast completion with small state. Running both through an unrestricted shared scheduler can let throughput-oriented work harm interactive latency.

Resource groups, admission, and placement can separate these priorities. A replica or analytical copy can isolate some read work while introducing freshness and replication costs. A hybrid engine must define how it handles version retention, background work, and contention between the two workloads.

A distributed analytical system can be excellent at scanning and grouping data while being a poor choice for a contended low-latency update invariant. A transactional system can support the report correctly while spending more resources than an analytical layout. These are workload consequences, not contradictions.

Compare three join plans with consistent units

Assume a fact input of 500 GB and a filtered dimension of 20 MB, both measured as serialized exchange payload in decimal units. Fifty workers already hold equal fact partitions. A broadcast plan sends the dimension to each worker, accounting for approximately one billion bytes, or 1 GB, of recipient payload if the sender itself is counted as a recipient for simplicity. Exact topology can save one local copy and add replication or framing overhead.

A repartition plan might exchange most of the 500 GB fact input plus the dimension. If roughly one fiftieth of rows are already at their destination under a uniform hash assignment, the expected remote fraction is forty-nine fiftieths. That gives about 490 GB of remote fact payload before protocol overhead. The broadcast plan appears attractive while the dimension remains small enough to fit per-worker memory.

Now suppose a cardinality-estimation error means the dimension is actually 8 GB after filtering. Broadcasting it to fifty workers creates about 400 GB of recipient payload and an 8 GB input representation at each worker before hash-table overhead. A plan selected from the 20 MB estimate can become infeasible. Runtime thresholds and a safe spill or replan path protect the engine from estimate errors.

A colocated plan can avoid most join exchange when both inputs are partitioned compatibly by join key and the needed replicas are available. Its apparent network advantage is purchased through data placement and maintenance decisions made before the query. Rebalancing, skew, and replicated dimensions can change those costs. There is no universally cheapest join independent of layout and input size.

Broadcast requires memory accounting, not just bandwidth

A dimension’s raw twenty-megabyte payload may become a much larger hash table due to buckets, pointers, alignment, variable-width allocation, and retained keys. If a hypothetical representation multiplies it by 2.5, each worker needs fifty megabytes for that state. Fifty workers retain 2.5 GB across the cluster for one join, excluding probe buffers and other operators.

If ten concurrent queries each construct the same table independently, the memory footprint multiplies again. Shared immutable build state can save work when snapshot, schema, authorization, and query semantics permit sharing. Sharing is not always safe across tenants or differing read timestamps. A cache key for execution state must include every semantic dependency.

A build-side memory limit should be checked during construction, before the process exhausts memory. Falling back to a partitioned or spilled join requires preserving rows already read and coordinating all participants. A local worker cannot unilaterally reinterpret an exchange protocol while peers continue sending under the old plan.

The build phase also affects first-row latency. Every probe worker may need sufficient dimension state before producing valid results. A small input can be cheap in total bytes while its acquisition is delayed by a slow source or remote scan. Critical-path timing and aggregate network volume are related but distinct measures.

Partial aggregation carries algebraic state

For AVG(amount), each worker sends a sum and non-null count. With one worker observing values 100 and 100, its state is (200,2). A second observing one hundred values of 1 has (100,100). Combining yields (300,102) and final average approximately 2.941176. Averaging the two local averages gives 50.5 and is incorrect because it discards weight.

Variance needs richer mergeable state and careful numerical formulas. A naive sum of squares can lose precision for large values with small variance. Exact distinct count needs enough information to resolve overlap across workers, while approximate sketches trade space and mergeability against an explicit error contract. The engine’s aggregate interface should define initialization, accumulation, merge, and finalization.

Grouping changes traffic according to distinct cardinality. Fifty workers each reading ten million rows but producing only one hundred region groups emit at most about five thousand partial groups before further combination. If every input row has a unique grouping key, partial aggregation may reduce almost nothing and can add hash-table overhead. The optimizer needs distinct-count estimates, and runtime measurements can reveal when the assumption was wrong.

Updates and retries require attention if partial results are reused. Adding the same worker’s partial state twice doubles the contribution. Task attempt identity and exchange semantics must ensure exactly one accepted contribution per logical partition, or use a replacement representation keyed by attempt/partition. A sum is mergeable but not idempotent under arbitrary duplicate delivery.

Global ordering and top-k need a proof

For a query requesting the ten largest rows by a deterministic total order, each worker can send its local top ten. Any row outside a worker’s local top ten has at least ten rows ahead of it from that worker and cannot belong to the global top ten. A final merge can select the answer from at most five hundred candidates across fifty workers.

The proof assumes the same ordering semantics everywhere and a total-order tie breaker if deterministic membership matters. WITH TIES, offset pagination, collation, null positioning, and duplicate semantics can require different candidate handling. An application wanting stable pagination should not rely on an incomplete ordering that lets tied rows appear arbitrarily.

Aggregation changes the problem. Selecting local top ten products by partial revenue can omit a product that is moderately popular on every worker but globally largest. The engine must combine product totals before safely applying a final top-k, unless a more sophisticated threshold algorithm proves pruning safe. Pushing down a limit blindly can return the wrong answer.

A globally ordered stream can merge sorted worker streams with a heap, retaining one current candidate per source plus buffers. If one source stalls and may contain the next smallest row, the merge can block. Output order creates a dependency on every relevant stream even when other workers have available rows.

Skew can be in data, work, or output

A hash repartition sends identical join keys to the same worker. If one customer owns 30% of fact rows, that worker receives a disproportionate share regardless of how many other workers exist. Hashing that same key again does not split it. A heavy-key strategy can isolate the key, replicate a smaller matching side, and divide its large-side rows among additional tasks while preserving output multiplicity.

A many-to-many join can inherently produce enormous output. If one key matches a million rows on each side, its cross-product has a trillion pairs. Better scheduling can distribute the work, but no execution engine can return those explicit pairs without paying for the result size. A small input key space is not necessarily a cheap join.

Work skew can occur with uniform row counts. One partition may contain expensive text, complex decompression, or a user function taking much longer per row. Progress metrics should include bytes and CPU as well as row count. Equal partitions by key count do not guarantee equal service time.

Adaptive splitting can improve balance when tasks are repeatable and boundaries are clear. The scheduler needs to prevent overlapping accepted output when a task is split or retried. An immutable range scan can use exact key boundaries and attempt identities. A mutating task requires a stronger transactional protocol before speculative duplicates are safe.

Exchange backpressure is part of query correctness

Every producer-to-consumer edge needs bounded buffers and an ownership protocol. A slow receiver should consume credits and eventually stop upstream production. Otherwise a fast scan can turn downstream CPU or network delay into memory exhaustion. Bounded queues make the resource limit explicit but can expose deadlocks if operators wait cyclically while holding incompatible resources.

worker scan -> local filter -> exchange sender
                                | bounded credits
                                v
                         exchange receiver -> join -> final aggregate

The caption identifies one network exchange between local pipelines. Credits bound outstanding payload, not merely message count, because rows can vary in width. A thousand tiny messages and a thousand ten-megabyte messages have radically different memory costs. Serialization scratch buffers and in-flight transport buffers also belong in accounting.

Cancellation propagates across the graph. If the final consumer disconnects, upstream workers should stop scanning, close streams, and clean temporary files. Late completions must not revive cancelled work or write into freed state. A query coordinator needs to distinguish successful task completion, retryable worker loss, cancellation, and permanent query failure.

Retries require a stable input snapshot if the query promises one. Rerunning a failed worker against newer data while retaining old results from other workers can produce a mixture that never existed. The engine can retain a read timestamp, restart the entire query, or provide a weaker documented contract. Fast retry is useful only while preserving query meaning.

Straggler mitigation should respect resource scarcity

Speculative execution starts another copy of a slow task and accepts one valid completion. It helps when delay comes from an unlucky worker and spare capacity exists. If every worker is slow because storage is saturated, duplicating scans can worsen the bottleneck and delay all tasks. The scheduler should distinguish isolated slowness from shared saturation.

The cost of speculation includes duplicate reads, network payload, retained snapshots, and cleanup. A losing attempt may finish after cancellation, so output acceptance must be fenced by task identity. If two copies both publish a partial sum, deduplication must prevent double contribution. Scheduling and aggregation semantics meet at this boundary.

Dynamic work assignment can reduce the need for speculation by keeping tasks smaller and balanced. Very small tasks increase setup, metadata, and RPC overhead. A sensible granularity amortizes those costs while keeping enough independent units to absorb variance. The choice follows measured task variability rather than a universal fixed chunk size.

A distributed report can use every core and saturate memory bandwidth while transactional queues grow. Per-query parallelism limits help only if aggregate admission is also controlled. Ten queries each respecting an eight-worker cap can still create eighty workers on a smaller machine.

Memory budgets must include global operator state, exchange buffers, spill files, and retained versions. An analytical replica can isolate some work, but its replication/apply path and freshness requirements remain. Running a report on a stale copy is acceptable only when the report’s contract allows that state.

ClickHouse’s distributed query facilities and CockroachDB’s SQL execution layer provide different implementation contexts for these mechanisms. The architectural comparison should trace a particular plan and its guarantees. A system optimized for reducing massive scan inputs can make different layout and scheduling choices from a system optimized for short contended transactions, even when both expose SQL.

Trace a complete distributed plan

Worker 1: fact scan -> filter -> local join -> partial(sum,count) --+
Worker 2: fact scan -> filter -> local join -> partial(sum,count) --+-> exchange
Worker N: fact scan -> filter -> local join -> partial(sum,count) --+      |
         ^ broadcast dimension                                         v
                                                              merge by group
                                                                     |
                                                              finalize average
                                                                     |
                                                              global order/output

The plan reduces data before the final exchange, but the earlier dimension broadcast remains a separate network cost. The partial aggregate carries sufficient state for correct weighted combination. The global output stage can become a single-resource ceiling if the result has many groups or a large ordered payload. Adding scan workers does not remove that final stage’s work.

If one worker fails after sending some partial output, the coordinator must know which logical input partitions those contributions covered. Retrying the whole worker and blindly adding its repeated partials would double count. Attempt-level acceptance or partition-keyed replacement state preserves exactly one logical contribution. This requirement is independent of whether the transport is reliable.

A complete query trace should report input bytes, filtered rows, join expansion, exchanged bytes, partial-group count, final-group count, spills, and time blocked on each exchange. The ratio between input and final output reveals reduction, while intermediate expansion identifies hidden cost. A ten-row final answer can still be expensive if a join generated billions of pairs before aggregation.

An execution plan is therefore a resource graph as well as a relational expression. Each edge carries bytes, ordering promises, progress state, and backpressure. Each node owns memory, CPU, and possibly spill storage. Explaining these properties makes the plan useful for capacity reasoning and failure analysis, whereas a list of operator names alone can conceal the work that dominates a distributed query.

Every exchange and operator needs an explicit resource budget.

Case connection and design conclusion

ClickHouse’s distributed engine and CockroachDB’s SQL-layer documentation expose different contexts for distributed query execution. The relevant comparison is the path taken by a particular query: partitions contacted, data moved, operators chosen, and guarantees preserved. A product-level throughput number does not substitute for that trace.

A distributed plan should explain locality, exchange volume, skew behavior, memory, cancellation, and the impact on other workloads. Lesson 32 explores how cached and maintained results can avoid repeating selected computations while introducing freshness and maintenance obligations.

Sources and further reading

Course index · Lesson plan