feat(data-plane): columnar engine on the Planner's Arrow operators; compile once; per-sample panes from columns

[zzylol]

Stack 5/5. Base: #831.

Why

The data plane's cost was the executor's row engine (Vec<Vec<Value>> batches, row clones, per-row grouping keys and accounting), not the summaries. ASAPPlanner main now runs every operator on Arrow batches, compiles plans that are Send + Sync, and offers StateColumnUpdater for per-sample state columns. This PR moves the backend onto that: compile once, build panes per sample from columns, feed raw windows as columns, and read rows only to write the answer.

What

• Re-pin to ASAPPlanner main 92bc1296.

• Compile once. The plan's ingestion side (Ingestion) and its query roots (QueryEngine) are compiled when the plan is installed and shared by every ingestion worker and query thread through Arc. The per-thread compile is gone.

• Raw samples to columns (series.rs). A raw-sample window's input batch is built from the samples' times and values and each series' cached cells: its series identity and label columns, dictionary-encoded. RawSampleSelector applies the Scan's matchers once per series. Query-time raw windows from the local raw buffer or remote read, kept panes, and the batch-build fallback all use it. No Sample and no row is made per sample.

• Ingestion per sample, no rows. For a definition whose group is a series (its group columns include the series identity):

  • the group comes from the cached per-series prefix;
  • StateColumnUpdater::update runs in place per sample;
  • at pane close, each worker seals its states and builds its part with Batch::try_from_columns, with the group texts, the pane's timestamp and the state column;
  • workers own disjoint series, so a pane is the workers' batches, stored in the SDS as Arc<Vec<Batch>> and shared by reference.

  Other shapes keep the plan's own build, now run once over every worker's samples of the pane, so a group spanning workers is never split. This replaces feat(precompute): update each open pane's summary state per sample #830's create_planner_accumulator path.

• Stored panes bind through Operator::source, which treats them as held outside the run, so their states are not charged to the query.

• Pane retention margin. Stored panes are kept one pane width longer than the plan's retention_ms: an evaluation at T runs once samples just after T have arrived, so the newest stored pane can be past T.

How

• SeriesWindow (one per raw-sample window, shared): it caches each series' cells or None (not selected) under a read-write lock, keyed by label set; workers and query threads look a series up once per run of its samples. batch(end, samples) builds the time, value and dictionary text columns and calls Batch::try_from_columns.
• PerSampleSpec (one per definition, shared) reads the SummaryAgg payload for the value, time and group columns, as the executor's lowering does. It maps each output text column to the group column of its name. It finds the output timestamp's offset from the pane end by running the plan's build once over one row.
• OpenPanes (per worker) numbers the worker's groups and holds one StateColumnUpdater per open pane; a pane's part is the touched groups, in group-key order as the batch build emits them.
• The engine compiles Arc<Ingestion> and Arc<QueryEngine> once in EngineHandle::spawn.

Deviation from the requested design. "Merge the shards' updaters" applies to groups that span workers. Every ingestion-time build the Planner selects for the examples is per series, so workers never share a group and there is nothing to merge. An empty KLL allocates its full item buffer, so aligning group numbers across workers would cost every worker an empty sketch per other worker's series per pane. Such a build keeps the plan's build over all workers' samples instead. The Planner placed no by-grouped build at ingestion for the queries tried (sum by (job) (count_over_time(...)), sum by (job) (sum_over_time(...)), sum(rate(...)), max by (job) (max_over_time(...))): without raw data they were all rejected as RawDataNotRetained.

Rows. grep -rn "to_rows\|Batch::try_new\|Vec<Vec<Value>>" data_plane/src finds one use: query.rs result_samples, which reads the root's result rows to encode the Prometheus JSON answer. Tests also read rows to compare panes (tests/per_sample.rs).

Transport, not rows. Raw samples still travel from the raw buffer and remote read to the query thread as (Arc<labels>, time, value) triples, which SeriesWindow encodes into columns per step.

Before this PR

#831: row batches throughout; each thread compiled its own plan; per-sample ingestion went through create_planner_accumulator, then rows; each worker's partial batch-build pane was concatenated (a group spanning workers would repeat). Panes were kept exactly the plan's retention, and the no-raw runs forwarded 160 of 481 evaluations of each example1 query.

After this PR

Arrow batches end to end, one compiled plan, per-sample panes built from columns, no forwards from retention. Unplanned queries are forwarded as before.

Verification

• cargo test --workspace passes, including e2e_examples (10: Examples 1, 3a, 3b with and without raw data, 4a, 4b, B3 kept results, and range queries over raw windows, stored panes and kept panes), kept, range, answers, raw_buffer. Every answer equals the Planner executor's query-time reference.
• per_sample::per_sample_panes_equal_the_batch_build: for Example 1 (counter rate, sum, 10 s panes), Example 3b (KLL, 1 min) and Example 4b (KLL, 10 min), on 1 and 4 shards, every pane built per sample equals the plan's own build over the same samples. Exact states are compared by serialized bytes, KLL by quantiles at every percent.
• New precompute::keeps_the_panes_of_an_evaluation_behind_the_newest_pane, failing without the margin: an evaluation at 10:00 still finds the 10-s pane it reads at offset 50 s after samples through 10:15. keeps_panes_for_the_plans_retention, range::every_step_reads_stored_panes and answers::forwards_whole_range_queries_a_step_cannot_answer move their evicted pane one width back.

Benefit harness (release, Prometheus 3.5.0, same session)

Same setup as #828: examples workload, 90 min of data, one sample per series every 15 s, AWS Fargate us-east-1 prices with measured operation costs. Prometheus is pinned to 4 cores and the data plane to another 4. "Prometheus alone" is the baseline run of the col experiment; "pr4" is #831's head, measured earlier. CPU is core-seconds over the whole run; latency p50 / p99 ms. Every evaluation in every col run meets its accuracy target. The 5 forwarded evaluations without raw data are Example 4b's first ones, before 60 minutes of panes exist (also forwarded before this PR).

s10

Run	Prometheus core-s	Data plane core-s	Data plane peak RSS GiB	rate p50/p99 ms	topk p50/p99 ms	p99[5m] p50/p99 ms	p99[60m] p50/p99 ms	forwarded
Prometheus alone	19.7	-	- (Prometheus 0.09)	10.0 / 19.3	10.1 / 19.7	21.2 / 127.2	110.0 / 144.5	-
pr4	5.4	36.6	0.48	24.0 / 45.0	21.3 / 42.9	86.2 / 141.3	564.7 / 689.7	0
col	5.6	16.3	0.15	10.4 / 12.5	10.8 / 12.7	38.0 / 41.2	126.8 / 139.0	0

s10-noraw

Run	Prometheus core-s	Data plane core-s	Data plane peak RSS GiB	rate p50/p99 ms	topk p50/p99 ms	p99[5m] p50/p99 ms	p99[60m] p50/p99 ms	forwarded
Prometheus alone	19.3	-	- (Prometheus 0.09)	9.8 / 19.2	10.1 / 19.3	20.6 / 33.2	100.7 / 134.3	-
pr4	10.1	31.7	0.30	25.3 / 48.1	20.3 / 25.3	73.9 / 87.4	96.2 / 133.3	325
col	5.8	20.0	0.34	10.2 / 21.0	7.8 / 10.0	62.1 / 74.6	99.0 / 121.1	5

s50

Run	Prometheus core-s	Data plane core-s	Data plane peak RSS GiB	rate p50/p99 ms	topk p50/p99 ms	p99[5m] p50/p99 ms	p99[60m] p50/p99 ms	forwarded
Prometheus alone	88.8	-	- (Prometheus 0.14)	44.4 / 72.9	45.8 / 70.8	94.7 / 134.3	432.9 / 550.8	-
pr4	27.6	152.6	1.76	90.5 / 131.6	88.8 / 118.8	385.3 / 435.3	2892.5 / 3287.0	0
col	28.3	68.4	0.32	41.3 / 64.5	32.6 / 54.2	138.3 / 191.6	506.1 / 546.6	0

s50-noraw

Run	Prometheus core-s	Data plane core-s	Data plane peak RSS GiB	rate p50/p99 ms	topk p50/p99 ms	p99[5m] p50/p99 ms	p99[60m] p50/p99 ms	forwarded
Prometheus alone	85.7	-	- (Prometheus 0.14)	42.7 / 73.7	43.7 / 73.4	98.6 / 146.5	361.3 / 579.4	-
pr4	47.1	133.6	0.97	105.6 / 202.9	64.4 / 96.2	320.8 / 368.4	470.4 / 594.5	325
col	29.2	88.5	0.78	49.4 / 88.1	28.5 / 42.8	254.0 / 297.1	399.3 / 482.4	5

Data plane + its Prometheus, against Prometheus alone (CPU core-seconds): s10 with raw 21.9 vs 19.7; s10 without 25.8 vs 19.3; s50 with raw 96.7 vs 88.8; s50 without 117.7 vs 85.7. The ASAP run's Prometheus still ingests every sample.

Where data-plane CPU goes now

perf record -F 499 -g on the data plane for a whole s50 run, with and without raw data (a separate run of each config, reusing its plan). Shares are of data-plane CPU.

By thread	s50, raw retained	s50, no raw
Query threads (4)	71.3%	62.4%
Ingestion workers (4)	9.1%	25.4%
HTTP runtime (Remote Write decode, raw-buffer reads, JSON)	19.6%	12.3%

By symbol group (self time)	s50, raw retained	s50, no raw
jemalloc (malloc, free, cache fills)	14.2%	12.5%
Kernel (page faults zeroing new pages, interrupts, network)	9.9%	21.7%
memcpy / memset / memcmp	8.2%	11.4%
Hash tables (ahash, hashbrown): executor and backend	11.0%	5.3%
Executor group tables (grouping keys into group ids)	8.2%	5.1%
Backend: series window, per-sample updates, raw buffer	7.0%	5.4%
JSON: answer encoding and series-identity decoding	6.1%	4.6%
Label-map hashing and comparison (SipHash, BTreeMap equality)	5.7%	3.9%
KLL kernels (update, merge, init)	2.2%	5.7%
Sorting	2.6%	3.3%
Summary merge and estimate operators	2.6%	1.7%
Protobuf and snappy (Remote Write)	2.5%	1.8%
Arrow kernels	1.6%	0.9%
Other	17.2%	16.0%

What remains:

• Without raw data, sketch allocation dominates. Every KLL is created with its full item buffer (KLL::init_internal). The zeroing shows up as memset and kernel page faults, about a third of the CPU, in every pane and every merge. A KLL that allocated lazily, or reused buffers, would remove most of it (an asap_sketchlib change).
• Label sets are hashed and compared as BTreeMap<String, String> (SipHash) in the backend's per-series caches and the raw buffer: 4–6%. Interning label sets to series ids at Remote Write decode removes this.
• Answers are encoded through rows: JSON writing, plus each answer row's series identity decoded from its JSON text, 5–6%.
• The executor's grouping (group tables, ahash) is the largest remaining operator cost.