← Back
Home › Systems › Databases and storage engines

query-engine

Plans a SQL statement and runs it in parallel over data it does not own, such as files, other databases, and streams.

Trino, DataFusion

Open the query-engine pagequery-engine Recursive-descent parsingcompilers: Hand-written top-down parser with one function per grammar rule; simple, fast, with full control over error messages and recovery.›Bloom filterdata-structures: Bit array answering "definitely not present" or "probably present" in about 10 bits per key for 1 % false positives; inserts are cheap, deletes impossible.›Data shuffledistributed: Hash/range repartitioning of intermediate data between workers (exchange operator)External merge sortsearch-sort: Sort data larger than RAM: write memory-sized sorted runs, then k-way merge them with large sequential buffers; terabytes in one merge pass.›Top-k selectionsearch-sort: Keep the k best of n items without sorting all n - bounded min-heap for streams, quickselect for arrays, threshold pruning to skip most items.›Cardinality estimationstorage: Row-count estimates from histograms, MCVs, sketches and sampling for the optimizerColumnar file formatstorage: Row groups, column chunks, pages, encodings and footer stats (Parquet/ORC style) (listed in Moved to the wiki)Columnar memory formatstorage: In-memory column buffers with validity bitmaps and offsets (Arrow style) for zero-copy exchangeFederated query planningstorage: Split one query across heterogeneous sources, pushing down what each can executeHash aggregationstorage: GROUP BY via hash tables with partitioning and spill to diskHash joinstorage: Build a hash table on the smaller input's join key and probe it with every row of the larger one; the linear-time workhorse for equi-joins.›Morsel driven parallelismstorage: Split pipelines into small morsels scheduled NUMA-aware across worker threadsPredicate pushdownstorage: Move filters and projections into scans and remote sourcesQuery optimizerstorage: Cost-based plan search: join ordering (DP/greedy), rewrites, access path selectionSort-merge joinstorage: Sort both inputs on the key, then advance two cursors in lockstep and emit matching runs; cheap on sorted inputs and good for range joins.›Vectorized executionstorage: Query operators pass batches of about 1-2K values per column instead of single tuples, amortizing interpretation overhead and enabling tight SIMD loops.›Zone mapsstorage: Per-block min/max (and sparse primary) indexes to skip data during scansFast hash functionsdata-structures: Non-cryptographic hashes built from 64×64→128-bit multiplies and xor-shifts; pick by key size, trust level and whether the hash is persisted.›Range partitioningdistributed: Cut the sorted key space into contiguous half-open ranges kept in a boundary map; split hot or big ranges, merge small ones, move ranges between nodes.›Buffer pools forzero-allocation I/Ohardware-os: Preallocate a fixed set of aligned I/O buffers and recycle them through a free list, so the hot path never mallocs, faults or unmaps. The bounded count doubles as back pressure.›K way mergesearch-sort: Merge k sorted streams with a heap or loser tree (merging iterators)Pattern-defeating quicksort(pdqsort)search-sort: Introsort successor: quicksort with adaptive pivots, linear on sorted and equal-heavy input, heapsort fallback; the unstable sort of Go and (until 1.81) Rust.›d-ary heapdata-structures: Array heap with d children per node (d = 4 typical): half the binary heap's depth, siblings in one cache line, faster priority queues.›HyperLogLogdata-structures: Counts distinct elements in kilobytes with about 1 % error by tracking the longest run of leading zero bits in hashed values per bucket.›Quantile sketchesdata-structures: Mergeable quantile sketches: t-digest, DDSketch, HDR histogramsReservoir samplingdata-structures: Uniform k-sample from a stream of unknown length (Algorithm R/L)Work-stealing dequeconcurrency: Per-worker deque: the owner pushes and pops at the bottom without contention, idle workers steal from the top; the core of Rayon, Tokio and Go.›Numa aware allocationhardware-os: Allocate memory and pin threads on the local NUMA node; first-touch and interleave policiesJoin order dynamic programmingstorage: DPccp and DPhyp enumeration of join orders with cost models; greedy fallback for many joins+ 3 morefsync, O_DSYNC and O_DIRECThardware-os: What fsync, fdatasync, O_DSYNC, O_DIRECT and F_FULLFSYNC really guarantee on Linux and macOS, and the exact call sequences for crash-safe files.›Memory orderingconcurrency: What Relaxed, Acquire, Release and SeqCst atomics actually guarantee, what they cost on x86 and ARM, and which one each pattern needs.›Epoch-based reclamationconcurrency: Threads announce a global epoch while touching shared lock-free data; garbage retired in epoch e is freed once the epoch reaches e + 2.›

How it is built

Architecture overview

A federated SQL query engine takes a SQL statement, plans it, and runs it in parallel over data it does not own: files in object storage, tables in other databases, streams. It has no storage of its own beyond caches and spill files. Trino describes the shape for the distributed case: one coordinator that parses, plans and schedules, and any number of workers that run the plan, with each data source plugged in through a connector ("an implementation of Trino's service provider interface", Trino docs). Apache DataFusion is the same pipeline as an embeddable Rust library, and Velox is only the execution half as a C++ library that "takes a fully optimized query plan as input" (Velox README). The idea that makes it fast is the split between a planner that rewrites the query using statistics and pushes work to the sources, and an executor that runs batches of columns through tight loops on all cores and exchanges partitions between machines. The two meet at a physical plan: a tree of operators with known partitioning.

Parts:

  • Parser — SQL text to a syntax tree — Recursive-descent parsing
  • Optimizer — rewrites the logical plan, orders joins, chooses access paths — query-optimizer (no page yet), relational-algebra (no page yet), predicate-pushdown (no page yet), subquery-decorrelation (no page yet), join-order-dynamic-programming (no page yet)
  • Statistics — row-count estimates that drive the choices — cardinality-estimation (no page yet), HyperLogLog, quantile-sketches (no page yet)
  • Federation layer — splits the plan across sources and pushes down what each can run — federated-query-planning (no page yet), columnar-memory-format (no page yet), partition-pruning (no page yet)
  • Executor — batch-at-a-time operators run on all cores — Vectorized execution, morsel-driven-parallelism (no page yet), Work-stealing deque
  • Operators — joins, aggregation, sort, top-k — Hash join, Sort-merge join, hash-aggregation (no page yet), Top-k selection, D-ary heap
  • Exchange — repartitions intermediate data between workers — data-shuffle (no page yet), Fast hash functions, Range partitioning
  • Spill — sort and aggregate more data than fits memory — External merge sort, PDQsort, k-way-merge (no page yet), Buffer pools and zero-alloc I/O
  • Data skipping — read less from the source — Bloom filter, zone-maps (no page yet), columnar-file-format (no page yet), Columnar storage
 SQL ─► parser ─► logical plan ─► optimizer ◄── statistics (row counts, NDV, sketches)
                                     │  push filters/projections/aggregates into scans
                                     ▼
                          physical plan: stages split at exchanges
        ┌────────────────────────────┼────────────────────────────┐
   stage 1 (scan splits)       stage 2 (partial agg)         stage 3 (final agg, sort/top-k)
   connector ► source          ◄── hash exchange ──►          ► client
   (Parquet, DB, API)          batches of ~1-2K values per column, Arrow-style buffers
        each stage: drivers = chains of vectorized operators, scheduled over cores (morsels)

How the parts fit together

Parser and logical plan

The front end is a hand-written recursive-descent parser (the alternative is a Pratt parser for the expression sub-grammar): one function per grammar rule, full control of error messages, no generator dependency. Its output is a syntax tree that is then bound to catalogs (names to tables and types) and turned into a logical plan: a tree of relational operators (scan, filter, project, join, aggregate, sort, limit) with no commitment on how each runs. DataFusion exposes this stage as a public API: you can build plans from SQL or from a DataFrame builder, and feed the same optimizer (DataFusion docs, query optimizer).

Optimizer

The block query-optimizer (no page yet) works on relational-algebra equivalences (relational-algebra, no page yet): selections pushed below joins, projections pushed to scans, constants folded, redundant joins removed. DataFusion documents this as a long list of logical OptimizerRules followed by physical ones such as join_selection (DataFusion docs, query optimizer). Three contracts matter:

  • Predicate and projection pushdown (predicate-pushdown, no page yet) is a contract with the source: the plan carries the filter, the connector reports which parts it can evaluate, and the engine re-checks only what the source declined. A source that returns only the needed columns and row groups is worth more than any operator tuning.
  • Join order is chosen by join-order-dynamic-programming (no page yet; DPccp/DPhyp with a greedy fallback for many joins) over costs derived from estimated rows. Cascades-style rule search (cascades-query-optimization, no page yet) as the alternative: it is more extensible, DP is simpler and finds the best order within its search space. A wrong order costs orders of magnitude, so estimation quality dominates.
  • Subqueries are unnested into joins (subquery-decorrelation, no page yet) so the join machinery can handle them.

cardinality-estimation (no page yet) feeds all of it: histograms and most-common values from the source, distinct counts from HyperLogLog sketches, quantiles from quantile-sketches (no page yet), and sampling. For a federated engine this is the weak point: each source reports different statistics, or none, so the optimizer must default to conservative guesses and prefer plans that are robust to a bad estimate (broadcast the small side only when it is known to be small).

Federation layer

federated-query-planning (no page yet) is what distinguishes this system from a warehouse. The planner asks every connector three things: what splits (units of parallel work: files, row groups, key ranges, shards) exist for this table, what operations can it push down, and what statistics does it have. Trino's coordinator "queries a connector for a list of all splits" and tracks which workers process which (Trino docs, concepts); a split is the contract of ownership between planner and scheduler. DataFusion's equivalent is the TableProvider trait, which lets a custom file format or non-file source plug in (DataFusion docs). Pruning of whole partitions (partition-pruning, no page yet) and files by statistics happens here, before any worker reads data.

Data crosses the source boundary in a columnar in-memory format (columnar-memory-format, no page yet; Arrow-style buffers with validity bitmaps and offsets) so that results from a database driver, a Parquet reader and an API all look the same to the executor, and so that operators exchange data without per-row conversion. Velox's vectors are described as "Arrow-compatible" (Velox README); DataFusion uses Arrow natively.

Executor

The key choice is vectorized execution over the tuple-at-a-time Volcano iterator (volcano-iterator-model, no page yet): each operator receives a batch of about 1-2K values per column plus a selection vector, so interpretation cost is paid once per batch and the inner loops are simple enough for the compiler and SIMD to speed up. The price is that batches must stay small enough for cache and that operators which need whole inputs (hash build, sort) are pipeline breakers that materialize.

Parallelism inside a machine is morsel-driven-parallelism (no page yet): the input is cut into small morsels, each worker runs a whole pipeline over one morsel, and idle workers take more. A work-stealing deque per worker is the scheduler structure; the alternative, thread-per-core (no page yet), pins data to cores and avoids stealing at the cost of skew sensitivity. Trino names the same levels: a task runs on a worker, a driver (a chain of operators with one input and one output) is "the lowest level of parallelism", and tasks consume splits (Trino docs, concepts). DataFusion calls its engine "vectorized, multithreaded, streaming" (DataFusion docs).

Operators

Hash join builds a table on the smaller input and probes with the larger one; it is the default for equi-joins and its cost is the build side's memory. Sort-merge join wins when inputs arrive sorted or the join is a range join, and it spills gracefully. The optimizer picks between them using the same size estimates, and a wrong pick shows up as either spilling or a slow probe. Aggregation is hash-aggregation (no page yet): a hash table of group keys with partial states, built per worker (partial aggregation), repartitioned by group key and merged (final aggregation). For ORDER BY ... LIMIT k the engine uses top-k selection with a bounded heap (D-ary heap) instead of a full sort, which makes k, not n, decide the memory. Hash keys use a fast non-cryptographic hash (Fast hash functions); a Bloom filter built from the build side and sent to the probe-side scan (a dynamic filter) removes non-matching rows before they are read or shuffled; DataFusion's docs refer to dynamic filters as a large speed-up for such joins.

Exchange

Between stages, data-shuffle (no page yet) moves batches: hash partitioning on the join or group key so equal keys meet on one worker, range partitioning (Range partitioning) for global sorts, broadcast for small tables, gather for the final result. In Trino, tasks write into an output buffer and consume from other tasks through an exchange client (Trino docs). The contracts are batch size (large enough to amortize network and serialization, small enough to keep pipelines streaming), back pressure (a full buffer pauses the producer), and partition count (too few leaves workers idle, too many produces tiny batches). Serialization uses the same columnar layout, so shuffling is mostly copying buffers.

Spill

When a hash table or sort buffer outgrows its memory budget, the operator spills: partitions of the hash table go to disk, and a sort writes sorted runs that are later merged. The sort path is external merge sort: sort memory-sized runs in memory (with PDQsort; radix sort is the listed alternative for fixed-width keys), write them, then k-way-merge (no page yet) them with large sequential buffers. Preallocated buffer pools keep the spill path free of allocation and give back pressure: a bounded number of I/O buffers. Spill makes a query slow instead of dead; it is the safety net, not the plan.

Data skipping

The cheapest row is the one never read. zone-maps (no page yet) hold min and max per block or row group, and columnar-file-format (no page yet; Parquet-style row groups, column chunks and footer statistics) lets the reader fetch only the needed columns and the needed row groups; the layout is described on Columnar storage. Bloom filters per column chunk skip row groups for point predicates. These live in the source's format, which is why the federation contract matters: the engine only benefits from what the connector passes down.

Data and control flow

One query:

  1. The client sends SQL to the coordinator (or calls the library). The parser builds the syntax tree; binding resolves names against catalogs.
  2. The optimizer rewrites the logical plan, pushing filters and projections to scans and choosing join order from statistics; the federation layer asks connectors for splits and pushdown capability, prunes partitions, and produces a physical plan cut into stages at exchanges.
  3. The scheduler assigns splits to tasks on workers. Time here is planning (milliseconds to seconds, longer with many joins or slow metadata calls to sources).
  4. Each task runs drivers: scan operators pull batches from connectors (decoding Parquet or calling a database), then filter, project, hash-join, partial-aggregate; output batches are partitioned and written to exchange buffers. Time goes to source I/O first, then decoding and hashing.
  5. Downstream stages pull partitions through exchange clients, run final aggregation, sort or top-k, and stream results to the client as they appear. Memory is bounded per operator; excess spills.

Dynamic filter (a join across sources):

  1. The build side (small) is scanned first and a Bloom filter or min/max range of the join key is collected.
  2. It is sent to the probe-side scan, which pushes it to the source or uses it to skip row groups.
  3. The probe side reads and ships far fewer rows. Skipping this step on a federated join is the common reason for a 100x slower plan.

Scaling

  1. Source I/O and decoding limit first. Levers: pushdown, partition and row-group pruning, columnar formats with statistics, local caching of hot files, and more connections per source within its limits (a federated source is often slower than the engine).
  2. Plan quality limits next: bad estimates produce bad join orders. Levers: statistics collection on the sources, sketches, hints, adaptive re-planning with run-time statistics.
  3. CPU limits when the data is cached: vectorized execution, morsel parallelism, SIMD, avoiding copies; then more cores per node.
  4. Network limits wide joins and aggregations: partial aggregation before the shuffle, broadcast of small sides, compression of exchange buffers, fewer stages.
  5. Memory limits hash joins and sorts: spill, smaller build sides, memory-aware admission control.
  6. Coordinator limits very many concurrent queries or huge plans: it plans and tracks every split, so it is a single point of load; scale by more or bigger coordinators, or by sharding workloads over clusters.

Failure modes

FailureSymptomCauseFix
Bad join orderone query takes minutes instead of seconds; huge intermediate resultsmissing or wrong statistics; independence assumption across predicatescollect stats, add hints, use dynamic filters
No pushdownthe whole table crosses the wireconnector or predicate not pushable (function, type mismatch)rewrite the predicate, expose the capability in the connector
Out of memoryworker killed or query abortedhash build or aggregation larger than the budget, many concurrent queriesspill, per-query memory limits, admission control
Skewed partitionsone worker runs long, the rest idlehot key in a hash exchange or joinsplit hot keys, repartition on a different key, broadcast
Slow source dominatesengine CPU idle, source CPU highfederated source cannot serve parallel splitslimit splits per source, cache or materialize, move the data
Small-file problemthousands of tiny splits, scheduler overheaddata lake with many small filescompact files, batch splits
Planning blow-upplanning takes seconds with 10+ joinsexhaustive join enumerationgreedy fallback beyond a join-count threshold
Type or null mismatch between sourceswrong results or errors in cross-source joinsdifferent semantics for decimals, timestamps, collationexplicit casts, connector type mapping tests
Stale metadataquery reads deleted files, misses new onescached file listings or partitionsshort TTL, table-format snapshots, refresh on query

Real implementations

namewhat it isrepolicencenotes
Trinodistributed SQL engine: coordinator, workers, connectorstrinodb/trinoApache-2.0stages, tasks, splits, drivers, exchanges as described in its docs
Apache DataFusionembeddable query engine in Rust on Apache Arrowapache/datafusionApache-2.0SQL and DataFrame APIs; TableProvider for custom sources; optimizer rules are public
VeloxC++ execution engine library (vectors, expressions, operators), no parser or optimizerfacebookincubator/veloxApache-2.0created by Meta; embedded in other engines

The alternative: a single-node analytical engine (embedded SQL over local columnar files) drops the coordinator, exchange and federation layers and keeps the vectorized executor; it is simpler and often faster when the data fits one machine. Pick the federated design when data lives in many systems you cannot move or when one query must use a cluster. A full warehouse (own storage with clustering and statistics) beats federation on planning quality because it owns its statistics.

Sources

Similar systems

Share the most building blocks with this one.

dataframe-engine 36 %batch-engine 36 %columnar-warehouse 34 %embedded-sql 29 %query-optimizer 28 %