Distributed Query Execution
Coordinator to workers to sources to partial results to merge — and the four places a query dies that a single-node engine never has.
Who needs this, what one row is, and why the obvious build breaks
Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.
What actually happens between submitting a SQL string and receiving the first row, when the work is spread across a coordinator and many workers?
The analyst waiting on a progress bar, and the platform engineer who has to explain why the query that ran yesterday failed today with a memory error. Both need the same thing: a mental model of where the work is, so that "it is slow" can become "stage 3 is waiting on one split".
Three nested units, and confusing them is the most common way people misread a plan. A stage is a piece of the plan separated from its neighbours by an exchange. A task is one worker's instance of a stage. A split is one bounded piece of input a task reads. Parallelism is bounded by splits; memory pressure is per task; the shape of the plan is per stage (Stages and Tasks).
Submit the SQL, watch the progress bar, and treat the cluster as a black box that goes faster when you add machines. When something fails, retry; when it fails again, ask for a bigger cluster. This works surprisingly often, because most queries are small and most clusters are over-provisioned relative to them.
One split holds far more data than the rest — a single enormous file, or a partition holding one dominant key — and the stage finishes when that one task does. Doubling the cluster changes nothing at all (Data Skew, Straggler Tasks).
- One split holds far more data than the rest — a single enormous file, or a partition holding one dominant key — and the stage finishes when that one task does. Doubling the cluster changes nothing at all (Data Skew, Straggler Tasks).
- The final aggregation, the
ORDER BYor theLIMITruns in a single task on the coordinator. A query that was perfectly parallel for four stages funnels into one process at the end and dies there (Query Optimizers). - A join whose build side was estimated at a few thousand rows is actually a few hundred million, because there were no statistics for the remote table. The broadcast becomes a network flood and every worker runs out of memory simultaneously (Broadcast Joins).
- The table is thirty thousand small files, so the coordinator spends the first minutes enumerating objects and computing splits before any worker does useful work — planning became the bottleneck (File Size and the Small-Files Problem).
- A worker is lost mid-query. Depending on the engine that is a retried task, a retried stage, or a failed query — and if intermediate results were held only in that worker's memory, the whole query restarts (Partial Failure).
- Twenty analysts submit at once. Each query is individually reasonable; collectively they exceed the cluster's memory, and the admission behaviour — queue, spill, or kill — decides whether this is a slow morning or an outage (Queueing: Why Systems Get Slow Before They Get Broken).
What is actually happening
- The coordinator parses the SQL, resolves names against the catalog, applies rewrites, chooses a plan, and cuts that plan into stages at every point where data must be redistributed. It then enumerates splits per source table and hands tasks to workers (How a Query Executes: Planner and Executor).
- A worker runs tasks. A scan task reads its splits, applies whatever filters and projections the reader can evaluate, and feeds the result into the operators above it in the same stage — pipelined, not materialised, so rows flow through filter and partial-aggregate without waiting for the scan to finish (Lazy Evaluation).
- Between stages sits an exchange: rows are repartitioned across workers by a hash of the join or grouping key, broadcast to every worker, or gathered to one. This is the same mechanism as a Spark shuffle and has the same cost structure — it is the point where data crosses the network (The Shuffle).
- Aggregation is normally two-phase. Each worker computes a partial aggregate over its own rows, the exchange redistributes by group key, and a final aggregate merges the partials. This is why
COUNTandSUMscale well andCOUNT(DISTINCT ...)does not — one merges cheaply, the other needs the values themselves (Aggregation: COUNT, SUM, AVG, GROUP BY, HAVING). - Memory is the binding constraint, and it is per task, not per query. When a task exceeds its budget the engine either spills the operator's state to disk and continues more slowly, or fails the query. Which of those happens for which operator is the single most important operational property of a given engine.
- Results stream back through the coordinator to the client. That makes the coordinator a real participant in execution rather than a pure planner, and the reason a query with a huge unaggregated result set can kill the one process everybody else depends on.
The path a query takes
Every distributed engine runs the same five-step arc, whatever it calls the steps: plan centrally, enumerate the input, execute in parallel, redistribute at each boundary, merge at the end. What varies is which step is your bottleneck, and the useful skill is being able to name it from evidence rather than from instinct.
Read the guarantees column below as the interesting one. Notice how little is promised at each hop: split enumeration promises the files the catalog listed, not the files that exist; partial aggregation promises a mergeable intermediate, not a correct answer on its own; the exchange promises delivery of rows to the right worker, and nothing about order.
The last stage is the one people forget. A query can be beautifully parallel through four stages and then funnel into a single task that sorts, limits or merges everything — and that final task is where memory failures concentrate, because it is the one place where the whole result exists at once.
- 1Parse and analyse
Turns SQL text into a logical plan, resolving every table and column against the catalog.
guarantees That names resolve and types are consistent with what the catalog claims.
fails by Succeeding against a catalog that is wrong about the schema — the plan is then valid and the reads are not (Metadata: Technical, Operational and Business).
- 2Optimise
Applies rewrites, pushes filters and projections toward the sources, chooses join order and join strategies.
guarantees A plan the engine believes is cheapest under the statistics it has.
fails by Choosing a broadcast from a default estimate when no statistics exist, which is the classic memory failure (Cost-Based Optimization).
- 3Split enumeration
Lists the files or remote pages for each source table and turns them into schedulable splits.
guarantees Coverage of exactly the files the catalog listed at planning time.
fails by Taking longer than the query itself when a table has hundreds of thousands of small objects (File Size and the Small-Files Problem).
- 4Scan tasks
Workers read splits, applying reader-level filters and projections.
guarantees Each assigned split is read once and only once into the pipeline above it.
fails by One split being much larger than the others, so the stage waits on a single task (Straggler Tasks).
- 5Partial aggregation
Each worker aggregates its own rows before anything crosses the network.
guarantees A mergeable intermediate for the aggregate — correct only once merged.
fails by Being impossible for holistic aggregates such as exact distinct counts, so all values must cross the exchange.
- 6Exchange
Repartitions rows by hash of the key, broadcasts a small side, or gathers to one node.
guarantees Rows arrive at the worker responsible for their key. No ordering.
fails by Hot keys sending most of the traffic to one worker — the same pathology as partition skew, one layer up (Data Skew).
- 7Final aggregate and sort
Merges partials, applies
ORDER BYandLIMIT.guarantees A complete, correctly ordered result over the planned snapshot.
fails by Running in one task and exhausting its memory on a result nobody meant to materialise.
- 8Result streaming
Returns rows through the coordinator to the client.
guarantees Delivery of the result set to a client that keeps reading.
fails by A slow client holding cluster resources open, or a large result set turning the coordinator into a memory bottleneck (Backpressure).
Four of these eight stages fail by succeeding slowly rather than by raising, which is why an engine with good stage-level metrics is worth more than one with good error messages.
Stages, tasks and splits
The three units are the vocabulary for every conversation about a slow query, and mixing them up makes the conversation useless. Parallelism is bounded by splits: a stage cannot use more workers than it has splits, so a table stored as one enormous unsplittable file is a single-threaded scan no matter how large the cluster (CSV, JSON and Their Limits).
Memory is bounded per task. Two tasks on the same worker share that worker's budget, which is why raising per-query concurrency can turn a stable workload into a spilling one without any query changing. And the plan shape is per stage — the exchanges are the only places where the engine's cost is about the network rather than about the data.
The diagram is the same picture as a Spark job with different vocabulary, and the correspondence is worth holding: stage boundaries are shuffle boundaries, tasks are per-partition units of work, and a straggler is a straggler in both (The Spark Execution Model). If you can read one, you can read the other.
1-- A: join first, then aggregate.2-- Every matching order row crosses the exchange before anything shrinks.3SELECT c.country, sum(o.revenue) AS revenue4FROM orders o5JOIN customers c ON c.customer_id = o.customer_id6WHERE o.order_date >= DATE '2024-03-01'7GROUP BY c.country;8 9-- B: aggregate to the join key first, then join.10-- The exchange now carries one row per customer, not one per order.11WITH per_customer AS (12 SELECT customer_id, sum(revenue) AS revenue13 FROM orders14 WHERE order_date >= DATE '2024-03-01'15 GROUP BY customer_id16)17SELECT c.country, sum(p.revenue) AS revenue18FROM per_customer p19JOIN customers c ON c.customer_id = p.customer_id20GROUP BY c.country;A good optimiser may derive B from A on its own — and may not, particularly when it has no statistics for customers. The point is not that B is always better; it is that the two forms move different amounts of data through the exchange, and the plan is where you find out which one you got (Reading EXPLAIN ANALYZE).
The four ways a distributed query dies
Single-node engines fail in one way: they run out of memory or they take too long. Distributing execution adds failure modes that are structural rather than incidental, and each has a different diagnosis and a different fix. Reaching for "a bigger cluster" is the correct response to exactly one of the rows below.
The trigger column is what you observe; the cause column is what is actually true. The gap between them is the entire difficulty of operating one of these systems, because the observable symptom of skew, of a bad broadcast and of an admission problem is identical: the query is slow.
Notice how many responses are layout or plan changes rather than capacity changes. That is the honest summary of this architecture: the engine gives you a lot of parallelism and almost no ability to fix data that was written badly (Physical Data Layout).
| Trigger | Symptom | Cause | Response |
|---|---|---|---|
| A stage sits at 99% for far longer than it took to reach 99%. | One task still running while every other worker is idle. | Skew — one split, or one key after the exchange, holds a disproportionate share of the data. | Find the key. Salt it, pre-aggregate before the join, or fix the file sizes that made the splits uneven. More workers changes nothing (Salting a Skewed Key). |
| Every worker fails with an out-of-memory error at the same moment. | The query dies seconds after a join stage begins. | A broadcast join whose build side was estimated small and is not — usually because the table has no statistics. | Compute statistics, or force a partitioned join. Treat the estimate, not the memory limit, as the defect (Broadcast Joins). |
| The query spends minutes before any worker starts. | Planning time dwarfs execution time; the cluster is idle throughout. | Split enumeration over a table with an enormous number of small objects, or a catalog listing that is itself slow. | Compact the table and adopt a format with manifests, so the planner reads metadata instead of listing storage (File Compaction). |
| The last stage fails, or hangs, after everything else succeeded. | A final sort or gather in a single task, holding the whole result. | An unbounded ORDER BY, a huge SELECT result, or a COUNT(DISTINCT ...) that could not be merged from partials. | Add a LIMIT, aggregate before ordering, or accept an approximate distinct where the question tolerates it. |
| The cluster is fine at night and unusable at 09:00. | Queue time dominates; individual queries look normal when run alone. | Concurrency exceeding cluster memory, with admission policy deciding who suffers. | Workload isolation and admission limits per group, plus moving scheduled jobs off the interactive window (Cost vs Freshness). |
| A query that has run daily for a year suddenly regresses after an upgrade. | Identical SQL, dramatically different runtime and plan. | The optimiser changed a decision — join order, broadcast threshold, a new rewrite. | Diff the plans, not the SQL. Pin what your engine lets you pin and treat engine upgrades as deployments with a rollback path. |
How to build it
Most important first.
- Reduce data as early in the plan as possible. Every row eliminated by the reader is a row that is never filtered, never hashed, never exchanged and never merged — which is why the pushdown lessons come immediately after this one (Predicate Pushdown, Projection Pushdown).
- Make splits even. Even splits come from even files, which come from compaction and sensible partitioning — a layout decision, made by the writer, that determines the parallelism the reader can achieve (File Compaction, Partition Cardinality).
- Aggregate before joining where the semantics allow. Joining two large fact tables and then grouping moves far more data through the exchange than grouping each side to its needed grain first (Grain: What Does One Row Represent?).
- Watch the build side of every join. If the small side is genuinely small, broadcasting it removes an exchange entirely; if the estimate is wrong, broadcasting it is the failure mode (Broadcast Joins).
- Set per-query memory limits and a concurrency limit that reflects reality, and make the rejection message say which limit was hit. A query that fails in ten seconds with a clear reason is worth more than one that degrades the cluster for an hour.
- Read the plan before optimising anything. Which stage is slow, how many splits it had, whether the filter is at the reader or above it, and where the exchanges are — all of that is printed, and guessing instead is how people end up adding workers to a coordinator-bound query (Reading EXPLAIN ANALYZE).
What this actually promises
Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.
- The query either returns a complete result over the snapshot it planned against, or it fails. Engines do not return partial results and call them answers — which is worth stating because pipelines that *wrap* a query sometimes do exactly that on timeout.
- Ordering is guaranteed only where the plan has an explicit final sort. Rows arriving in a plausible order from a parallel scan is a coincidence that will change with cluster size (Ordering Guarantees: Four Levels, Four Prices).
- Task-level retries are transparent for pure scan work and are not transparent for stateful operators whose partial state was lost. What is retryable is an engine property, not a general one.
- Nothing guarantees that two runs of the same query cost the same. Split assignment, cluster occupancy and which node holds a cached file all vary, and none of them is under the query author's control.
- There is no cross-table snapshot unless a table format provides one. A join across two tables can read each at a different commit (Open Table Formats).
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- The check that matters for execution is a result-stability check: run the same query twice against the same table snapshot and assert identical results. It catches non-deterministic plans, an unstable
LIMITwithoutORDER BY, and aggregation over a table that is being written concurrently. - It misses every case where the query is deterministic and wrong — a fan-out join, a filter at the wrong grain, a
SUMover a table that already aggregates. Determinism and correctness are unrelated properties (The Pipeline Succeeded. The Data Is Wrong.). - For pipelines built on engine queries, add a row-count assertion against the expected grain at the output. A distributed query that silently read fewer splits than it should have looks exactly like a quiet business day (Reconciliation).
- Interactive execution is what this architecture buys: no load step, no materialisation, an answer computed from current files at query time. The freshness ceiling is the writer's commit cadence, not the engine.
- The latency floor is planning plus split enumeration plus at least one exchange round. For small queries this fixed cost dominates, which is why a distributed engine is often slower than a single-node one on a small table (DuckDB Concepts).
- Under concurrency, queue time is part of freshness from the consumer's point of view. A dashboard that refreshes on a schedule and waits ten minutes in admission is stale for reasons that have nothing to do with the pipeline (Freshness Monitoring).
- Adding a column changes plans in one specific way:
SELECT *in a view or a downstream model now reads more, everywhere, without any query being edited (Projection Pushdown). - A type change on a partition column is more disruptive than it looks, because pruning depends on the engine matching the predicate's type to the partition value's type. A column that becomes a string still queries — and stops pruning (Predicate Pushdown).
- Engine upgrades change plans. A cost model change or a new rewrite can turn a broadcast into a partitioned join or vice versa, and the query text is identical on both sides of the upgrade. Plan regressions are a real deployment risk (Query Optimizers).
- A read query needs no recovery: retry it. This is one of the few places in this domain where the answer really is that simple, because the query left nothing behind.
- A query that writes needs the same protection as any pipeline step — write to a staging location, validate, publish atomically — because a failed distributed write leaves some tasks' output committed and others' not (Atomic Publish).
- When a query fails for memory, the recovery is a plan change, not a retry: reduce the input with a better predicate, aggregate earlier, disable the broadcast, or split the work by partition and union the results (Incremental Processing).
What can go wrong
- One straggler task holding a stage while every other worker sits idle (Straggler Tasks).
- Coordinator memory exhaustion from a large final result, a huge plan, or split enumeration over a table with far too many files.
- Broadcast join based on a wrong estimate, failing every worker at once.
- Spill turning a fast query into an extremely slow one — success that is indistinguishable from a hang to the person waiting.
- Cluster-wide starvation from concurrency, where every individual query is reasonable and the aggregate is not (Queueing: Why Systems Get Slow Before They Get Broken).
- The mitigation failing: a per-query memory limit set low enough to protect the cluster also kills the legitimate monthly aggregation, at night, into a retry loop nobody watches.
- "Add workers and it gets faster." Only if the work is evenly divisible and the bottleneck is worker CPU. Skew, a coordinator-side merge and a non-pruning scan are all immune to more workers (Amdahl's Law).
- "The query is slow because the cluster is small." Check queue time first. A cluster that is fast when idle and slow at 09:00 has a concurrency problem, and buying capacity for it is the most expensive possible fix.
- "Stages equal steps in my SQL." Stage boundaries are exchanges, not clauses. One SQL statement can become a dozen stages, and a join in the text may vanish into a broadcast that has no stage of its own.
- "It failed on memory, so we need more memory." Sometimes. Far more often the plan is reading or moving data it did not need to, and the memory error is the symptom furthest from the cause (Predicate Pushdown).
Operating it
- Per stage: input rows, output rows, splits, wall time and the slowest task. The ratio of slowest task to median task is the skew signal, and it is the first number to look at (Data Skew).
- Bytes exchanged per query. It is the network cost, and it distinguishes a query that is scanning too much from one that is moving too much (The Shuffle).
- Spill events and peak memory per task, over time. Spill is the early warning that precedes the memory failures by weeks.
- Queue time versus execution time, per query and per user. When queue time dominates, the fix is admission control and workload isolation rather than query tuning (Capacity Planning: Traffic to Machines).
- At 10x data, well-partitioned queries scale nearly linearly because they read proportionally more splits. Queries that never pruned scale linearly too — from a bad starting point (Partition Pruning).
- At 100x, the coordinator becomes a scaling limit of its own: planning time, split counts and result gathering all grow with the table rather than with the answer.
- At 10x concurrency, the constraint moves from throughput to admission. The question stops being "how fast is one query" and becomes "which queries are allowed to run at the same time" (Little's Law as Working Intuition).
- Bytes read from storage, decided by pruning and projection, is the biggest lever and the one furthest upstream (Scan Cost).
- Bytes exchanged is second and is a property of the plan: broadcast versus partitioned join, one-phase versus two-phase aggregation, and whether the grouping happened before or after the join.
- Cluster hours held, including idle time. An always-on cluster sized for the peak is paying for the peak continuously, which is the structural argument for elastic or single-node execution (Compute Waste).
- Coordinator work — planning, split enumeration, result gathering — is invisible on most cost dashboards and is exactly what a small-files problem inflates.
- Distributing execution buys the ability to query data that does not fit on one machine, and costs a coordinator, an exchange, a scheduler, per-task memory management and a class of failures — skew, stragglers, broadcast blowups — that simply do not exist in one process (Why Eight Cores Give You Four and a Half).
- Pipelined execution buys low time-to-first-row and costs restartability: with little materialised between operators, losing a worker can mean redoing more work than a checkpointing engine would.
- Memory limits buy cluster stability and cost the tail of legitimate large queries. There is no setting that protects everybody and blocks nobody.
Query stage explorer
Change an input and watch which number moves — and which one does not. Everything here comes from a model in this repository, not from a measurement.
SELECT country, sum(revenue) FROM events WHERE event_date = '2026-08-25' AND revenue > 0 GROUP BY country
Where this applies
Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.
- GENERALCoordinator, workers, splits, stages, exchanges and two-phase aggregation describe every distributed SQL engine in wide use. What differs is naming — stage versus fragment, split versus task — and it differs enough to make documentation confusing.
- ENGINE-SPECIFICWhether a lost worker retries a task, retries a stage or fails the query, and whether a hash aggregation spills to disk or dies, are engine-level design choices. An engine that spills degrades gracefully and hides problems; one that fails fast surfaces them and interrupts people.
- SIMPLIFIEDDescribed as a single coordinator planning and a homogeneous worker pool executing. Real deployments have separate planning and dispatch roles, worker groups with different resource shapes, and caching tiers between worker and object storage.
Where the depth lives
This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.
- — Distributed Systems owns what happens when a worker is lost mid-query: what the coordinator can assume, how it detects the loss, and why "retry the task" is only safe for work with no external effects.