Skip to content

Execution and parallelism

collect() lowers the optimized plan and runs it on a block-parallel pipeline: workers claim block-sized units of work from the source, stream each block through the fused row-preserving prefix, and push it into the query's sink — the stage that accumulates the answer. There is no per-stage materialization between a scan and its sink; a block flows source-to-sink in one pass.

The pipeline

Lowering folds every run of row-preserving nodes — filter, select, with_columns — into one fused prefix. Per block, the prefix evaluates the scan predicate first and compacts survivors once; columns arrive as deferred handles, so a payload only decodes when an expression actually touches it, and a predicate a block's zone statistics resolve to all-false skips the block without faulting its pages. A bare column passing through untouched is a reference-count bump, never a rebuild.

Expressions run as flat programs with common subexpressions shared: lowering strips aliases, memoizes structurally identical subtrees, and — in aggregation sinks — merges every aggregate's input into one program per block, so a subexpression two aggregates share evaluates once.

Workers claim blocks either dynamically (work-stealing, for order-free sinks) or as contiguous ranges when the sink's semantics need data order — first/last aggregates, sorted-run grouping, window barriers. Worker count and thresholds are read fresh at each stage boundary; a runtime change never lands mid-stage.

Sinks and strategy graduation

The sink family covers table assembly, streaming top-k, global aggregates, grouping, sorts, the window barrier, join probes, and native-file writers. Where more than one strategy can serve a shape, the engine picks by measured evidence, and every strategy is test-pinned to produce identical results:

  • Grouping is the deepest family. The common case is a vectorized columnar table — batch-hashed keys, monomorphized update kernels per aggregate. Small coded key domains direct-address by composite code with no hashing at all; a single int key with tight value bounds direct-addresses by value; a key that block metadata proves clustered (sorted blocks, chained bounds — a symbol-sorted table) folds runs in place with no table whatsoever; a key an ingest sketch says is high-cardinality scatters to per-partition tables so the merge stays partition-local. The row-oriented engine remains for the shapes the vectorized set declines (heap-backed side states such as median/n_unique).
  • Text stays coded. Dictionary-coded columns group, compare, and join in the code domain; per-code hashes come from a query-wide memo, and results re-seal as coded blocks — strings materialize only when a consumer finally looks at them.
  • Joins build the right side, then stream the left through a probe chain. Every probe consults a compact membership filter before the hash table — rows the build side cannot match never touch it, and a filter that stops paying for itself retires. Interior dimension trees materialize into one composite build where that provably reduces probes. Semi and anti joins never materialize the join at all. Asof joins verify sortedness from block metadata, then run a monotone per-group merge.
  • Sorts order by packed radix keys when the whole key tuple fits an order-preserving word, and fall back to a stable comparison sort otherwise; a single ascending key whose blocks prove already-ordered returns its input untouched. Gathers permute each column once, in parallel.
  • Windows are a barrier by nature — a window value reads rows outside any block — so the sink assembles its input, then evaluates partitions with its own parallelism, recognizing clustered partition keys as contiguous runs.

The output contract

Group and join output order is unspecified. Rows are complete and correct; their order is not part of the contract — the in-memory, parallel, and spilled paths are each deterministic but need not agree, and not promising an order is what lets them skip global reorders. Sort output is ordered by definition, stably, with null smallest in every kind. Within a group, rows fold in input order, so first/last are exact. Integer results are identical across worker counts by construction; float sums and means may differ in final bits because float addition reassociates under parallelism — the one documented carve-out. Float keys group and join by bit pattern; null is a groupable value but never matches as a join key.

Cancellation and progress

A per-query context is checked at block boundaries: a block's work always completes once started, so cancelling leaves a prefix of whole blocks, never a torn one. Progress counts blocks and rows as they stream.

Out-of-core execution

Queries are not limited by memory. The engine derives a budget from the effective total — inside a container, the container's allowance — and every stateful sink consults it:

  • Map-only pipelines don't need one. scan → filter/derive → collect_to_native streams from file mapping to output file in constant memory.
  • Sorts spill runs — budget-sized sorted chunks on disk, k-way merged; bit-identical to the in-memory sort, stability included.
  • Grouping partitions — rows hash-partition to disk by key, each partition aggregates independently, no cross-partition merge.
  • Joins go partition-pair — both sides partition their row indices; each pair joins in memory, one at a time.

Spill files are ordinary native tables in a per-query temporary directory, removed however the query ends. Mmap-backed scan payloads never count against the budget — the kernel pages them.