Physical operator library performance: columnar, batched, parallel and efficient basic operations #778

Problem, by example

The executor (asap-executor) is ASAPPlanner's physical operator library. Runtimes such as ASAPQuery-backend run every installed DAG node with it, so its speed is the runtime's speed. #762 made every operator columnar, but per-operator efficiency is still far from a mature engine such as DataFusion.

A concrete case: sum by (job) (rate(http_requests_total[1m])) over stored panes, 1,000 series per metric, one core, the backend's in-process benchmark:

For the work involved (grouping about 1,000 rows into a handful of groups), these numbers are high. Profiles of the data plane show the remaining time in four places:

• group-table hashing (13–19%);
• allocation (12–14% with jemalloc, more with glibc);
• per-row scalar code;
• concatenating stored panes before each query.

Sketch kernels themselves are 2–6%.

What to improve

1. Column-wise execution everywhere. No operator decodes rows. The remaining row paths are listed in asap-executor: a columnar physical operator library (state columns, per-sample updaters) for runtimes executing the DAG #762's follow-ups (the join's error replay, literal batches) and in the backend's encoding at its edges. Expressions evaluate as vectorized Arrow kernels over whole columns, not per row.
2. Batch execution. Operators consume and produce bounded batches (asap-executor: a columnar physical operator library (state columns, per-sample updaters) for runtimes executing the DAG #762 uses 1,024-row chunks for memory checks), streaming end to end instead of materializing whole inputs. That covers, for example, merging N stored panes without first concatenating them, and pipelining Scan → Filter → Aggregate.
3. Parallelism. Compiled plans are Send + Sync since feat(executor): remove the row form of Batch, Send + Sync compiled plans, no charge for stored states (#762, 4b) #768. The executor can use that:
   • partitioned aggregation (hash-partition by group, aggregate per partition, merge);
   • parallel merge of stored panes;
   • independent DAG branches on separate threads, under the runtime's thread budget.
4. Efficient basic operations. Even sum matters:
   • Use SIMD-friendly loops or Arrow compute kernels for sum, count, min, max and comparisons.
   • Specialize grouping for one-column or dictionary keys (group by dictionary index, no hashing).
   • Avoid per-row branching on null bitmaps when a column has no nulls.
   • Keep compensated (Kahan) summation where PromQL semantics require it, and use the fast path elsewhere.
   • Reuse buffers across batches.

How to measure

• A benchmark next to DataFusion. DataFusion is already a dependency of the SQL frontend.
  • Run the same exact aggregation queries (filtered sum/count/count(distinct) by group, top-k, joins) on the same Arrow data in both engines, and in the executor through a planned DAG.
  • Report absolute times per query and the gap.
• The backend's in-process benchmark: pane construction per sample, stored-state queries, and data-plane CPU in the benefit harness.

Acceptance

• A checked-in benchmark that compares the executor with DataFusion on a fixed query set, with results recorded in the PR that adds it.
• Each improvement lands with before/after numbers from that benchmark, and the existing executor tests still pass.

Related: #762 (columnar operator library), #761 (Stage 3 CPU under-prediction), #771 (measured costs).