perf: parallelize serial stages around group-by (binary aggs, OP_IF, filter/where/gather)

[ser-vasilich]

perf: parallelize the serial stages around group-by (binary aggs, OP_IF, filter/where/gather)

Problem

On a 48-core machine the canonical TAQ benchmark (medium NYSE dataset, quote = 296.6M rows) showed rayforce group-by queries scaling only 2.1–2.3× from 1 to 48 cores. Profiling the full query pipeline showed the group-by kernel itself was fine after per-worker accumulation — the wall-clock was dominated by serial stages around it:

• binary aggregates (pearson/cov/wavg/wsum) were not DA-eligible → fell to slower paths;
• OP_IF projections went through exec_if_selected (serial per-row id-scaffolding) even when no WHERE was present;
• bitmap→index materialization in exec_filter / sel_compact was serial;
• the where builtin and at col idx gather were serial;
• ray_pool_dispatch was being fed chunk counts, so 1500+ chunks morselized into ~2 tasks.

What changed (8 commits + 2 dev merges)

src/ops/group.c — binary aggregates via the DA (dense-array) path: per-worker accumulators extended with co-moments (Σx, Σx², Σy, Σy², Σxy), finals computed in emit_agg_columns. Two merge bugs fixed along the way: Σx for integer x-columns is stored as double but was merged through the int64 union member (garbage like 1.3e305 at high core counts); the da_merge_fn path didn't merge the PAIR co-moment arrays at all (wrong pearson with ≥1024 slots). Eligibility is gated: y-column must be FP or non-nullable integer — nullable-int y falls back, since DA reads raw values and would sum null encodings as data.

src/ops/pivot.c — OP_IF: (1) the eager elementwise fill is parallel (type-switched morsel fill; SYM LUT warm-up stays serial before dispatch; STR output stays serial); (2) trivial-branch ifs (scan/const through alias/materialize) are rerouted from exec_if_selected to the eager path, gated by an out-type whitelist — exec_if_selected fires even without a WHERE clause and was the hidden per-row serial cost in Q22-25 projections.

src/ops/filter.c — exec_filter and sel_compact: bitmap→match_idx build parallelized with 3-phase compaction (per-chunk popcount → serial exclusive prefix sum → per-chunk fill at disjoint offsets), one task per chunk via ray_pool_dispatch_n. Chunk count capped at 1024 (the initial task-ring capacity), so ring growth can never fail and silently drop tasks.

src/ops/builtins.c / src/lang/eval.c — where builtin parallel with the same 3-phase pattern (keeps the word-at-a-time zero-skip in the emit phase); gather_by_idx parallel with a ray_parallel_flag reentrancy guard (existing callers may already be inside a dispatch); null-bit propagation stays serial.

All parallel paths share the same gates: pool && pool->n_workers > 0 (at -c 1 the pool exists with 0 workers), len >= RAY_PARALLEL_THRESHOLD, serial fallback on any allocation failure.

Results

48-core box, canonical queryEngines.sh, medium TAQ, min of 3 runs, both binaries built from the same dev base (8f8033c):

Query	dev c48 (ms)	branch c48 (ms)	speedup	scaling 1→48 before	after
Q22 (wavg by ex)	3932	1443	2.72×	2.1×	5.6×
Q23	3731	1405	2.66×	2.2×	5.7×
Q24 (by sym,ex)	6180	2293	2.70×	2.3×	5.9×
Q25	5479	1770	3.10×	2.2×	6.7×

Side wins from the filter/where/if work: Q48 2.28×, Q42 1.69×, Q47 1.63×, Q7 1.49×, Q39 1.29×, Q8/Q16/Q17/Q19/Q41 1.2–1.3×.

Single-core: no regressions — branch ≤ dev at c1 across all 53 queries (e.g. Q22: 8052 vs 8216 ms), thanks to the n_workers > 0 gates.

Correctness

• make test: 3639/3639 on a clean build.
• wavg/pearson results verified against numpy to full precision at c1/c4/c24, including FP-y columns with NaNs and integer x-columns.
• Edge cases that can't be handled safely (nullable-int y, STR outputs, mixed-type if branches, reentrant dispatch) fall back to the existing serial paths via explicit gates.

Known remaining gaps (out of scope here)

• duckdb is still faster on Q22-25 at high core counts (530–1864 ms): its advantage is fusing the filter/projection expressions into one pass, while rayforce materializes each op. Closing that needs expression fusion into the compaction pass — architectural.
• Window functions parallelize across partitions only (window.c Pass 3); low-cardinality partitioning (e.g. by exchange = 19 parts) caps their scaling.

🤖 Generated with Claude Code

https://claude.ai/code/session_01JZwH1gpAeLzDX2bxc4jz6s