Topic 11 — Query Engines II: Execution Models

Volcano vs vectorized vs compiled — the defining performance battle of modern analytics. Topic 10 chose the plan; this topic is about how fast you can RUN it. The gap between tuple-at-a-time and vectorized execution is one to two orders of magnitude, and you will measure it yourself.

flowchart LR
    P[physical plan] --> M{execution model}
    M -->|"Volcano '90"| V["next() per TUPLE<br/>interpret per tuple"]
    M -->|"X100 '05"| X["next() per VECTOR<br/>interpret per 1-2K batch"]
    M -->|"HyPer '11"| H["compile plan to<br/>machine code, no next()"]
    V --> C[same results,<br/>10-100x apart]
    X --> C
    H --> C

The problem, measured (bench lane 1, provided — runs today)

cargo run --release --bin exec_bench — 50 M rows, SELECT k, SUM(v) WHERE f < t GROUP BY k, best of 3:

selectivity      volcano
        5%      0.386 s     129.4 M rows/s
       50%      0.484 s     103.3 M rows/s
       95%      0.669 s      74.7 M rows/s

103 M rows/s is the tuple-at-a-time ceiling on this machine — and notice which way it moves. Cost rises monotonically with selectivity, because in a Volcano pipeline every row that survives the filter pays the full per-tuple interpretation bill on its way to the aggregate: a virtual call per operator per row. The filter is not the expensive part. Passing it is.

Put that next to topic 12’s measurement of the same machine — ~50 GB/s of sequential u64 fold, about 6.2 G values/s — and the gap is 60×. That gap is what vectorized execution and compiled kernels are competing to close, and the interesting question is not whether they close it but what the remainder consists of once they do. Predict the split before you implement either lane.

1. The Volcano (iterator) model

Every operator implements open() / next() / close(); next() returns ONE tuple. Elegant: operators compose arbitrarily, demand-driven, bounded memory.

 Project.next()
   └─ calls Agg.next()
        └─ calls Filter.next()      per-tuple costs, PER TUPLE:
             └─ calls Scan.next()   - virtual call (indirect branch) x depth
                                    - interpretation of the expression tree
                                    - tuple is gone from registers between calls

Postgres still runs this (ExecProcNode — a function pointer per node), and for OLTP it’s fine: a point query touches 3 rows; who cares about per-tuple overhead. The disaster is analytics: 100M rows × 5 operators × ~20 ns of interpretation overhead = minutes spent NOT computing.

2. Vectorized execution (MonetDB/X100 → DuckDB)

Same iterator shape, but next() returns a BATCH (DuckDB: DataChunk, 2048 rows). All the per-call overhead amortizes over the batch, and the inner loops become tight for over columnar arrays — the compiler auto-vectorizes, the prefetcher streams, branches disappear.

 per-TUPLE model:   overhead × N_rows
 per-VECTOR model:  overhead × (N_rows / 2048)  +  tight loops over arrays

Two supporting tricks (both in DuckDB, both worth stealing for M11):

• Selection vectors: a filter doesn’t copy survivors — it produces an index array sel[] over the same vectors. Downstream kernels take (data, sel, count). Zero copies until materialization is forced.
• Vector type flags: a vector can be FLAT, CONSTANT (one value — arithmetic with a constant never expands it), DICTIONARY (selection over a dictionary — compressed data flows through the engine). Kernels dispatch on the combination.

Why 1–2K rows? Big enough to amortize call overhead, small enough that a chunk’s working set stays in L1/L2 between operators. It’s the cache ladder of topic 0 turned into an engine design parameter.

3. Compiled execution (HyPer)

Skip interpretation entirely: fuse each PIPELINE (chain of operators between materialization points) into one tight loop and JIT it — the tuple stays in CPU registers from scan to sink.

 for (row in fact_table)              // one compiled loop = whole pipeline
     if (row.f < 50)                  // filter: a branch, not an operator
         ht[row.k] += row.v;          // agg: an add, not a next() chain

VLDB’18 (“Everything You Always Wanted to Know…”) raced the two champions: roughly equal on aggregation-heavy work; compilation wins complex expressions and tight OLTP; vectorization wins compile-time (ms vs 100s of ms), profiling, and adaptivity. DuckDB chose vectors partly for engineering reasons — no LLVM dependency, debuggable C++ kernels. Topic 19 revisits compilation; M11 goes vectorized.

4. Morsel-driven parallelism (SIGMOD’14)

How to parallelize pipelines: break the input into MORSELS (~100K rows), workers PULL morsels dynamically instead of getting static partitions.

• NUMA-aware: a worker prefers morsels on its socket.
• Elastic: skew doesn’t strand workers (no “thread 3 got the hot partition”); a slow morsel just means that worker pulls fewer.
• DuckDB: source operators hand out row-group-sized work units (122880 rows = 60 vectors); MaxThreads on the source caps fan-out. polars-stream literally names its unit Morsel and pairs it with a sequence token so order-sensitive sinks can reassemble.

        ┌ morsel queue ┐
 scan:  [m0][m1][m2][m3][m4]...
          ▲    ▲         ▲
        w0 pulls, w1 pulls, w2 pulls   (dynamic — no static split)
 each worker runs the WHOLE pipeline on its morsel:
 scan → filter → probe → partial agg (thread-local HT)
 then: combine partial HTs (the sink's Combine/Finalize phase)

5. Hash joins and aggregation, vectorized

The two operators where analytics time actually goes.

• Hash join (DuckDB join_hashtable.cpp): build side materializes into row-format tuple data, partitioned by hash radix; the hash table stores 8-byte entries = pointer + SALT bits (topic 2’s bit-smuggling: compare salt before chasing the pointer — most misses never touch the tuple). Probe is vectorized: hash 2048 keys, gather 2048 buckets, compare salts, chase survivors via selection vector.
• Hash aggregation: same skeleton, plus two-phase: threads build thread-local partial HTs (no contention), then radix-partitioned merge (RadixPartitionedHashTable). DataFusion’s GroupedHashAggregateStream interns group keys → dense group index → aggregate states live in flat columnar arrays indexed by group id (not per-group heap objects).

Experiments (experiments/)

One query, three engines — SELECT k, SUM(v) WHERE f < t GROUP BY k:

1. volcano.rs — PROVIDED: tuple-at-a-time iterators, dyn dispatch. The honest 1990 baseline.
2. vectorized.rs — YOU implement: 1024-row batches, columnar arrays, selection vectors, group-by into a flat array (k is dense).
3. kernels.rs — YOU implement: branchless/SIMD-friendly single-pass fused kernel (what a compiled engine would emit for this pipeline).

cargo test checks all three agree with a scalar oracle; cargo run --release --bin exec_bench prints rows/s for each — the gap IS the lesson. Predict the two ratios in notes.md first.

Reading guides

Further references: “Photon” (SIGMOD 2022) — Databricks’ vectorized C++ engine, notable for why not compilation: easier to build, debug, and roll out than codegen at their scale; “Velox” (VLDB 2022) — Meta’s reusable execution library (the executor as a component, like Calcite for topic 10); “GAMMA” (1986) — where exchange-operator parallelism came from, if you want the pre-history of §4.

Capstone M11

Vectorized runtime for the graph engine (reference mirror: runtime/batch.rs, vectorized.rs, eval.rs):

• batch type: fixed-capacity columnar chunk (node ids, properties) + selection vector — pick the batch size by measuring, not by copying DuckDB’s 2048
• operator pipeline: source (label scan) → expand → filter → project, each fn next(&mut self, out: &mut Batch)
• expression eval over batches (property predicates) — no per-row dyn dispatch
• the FalkorDB angle: Expand over a sparse matrix IS a vectorized kernel already (one GraphBLAS call per batch of source nodes); decide where matrix ops end and row-ish batches begin
• bench: M11 runtime vs M10’s naive interpreter on the same plan