[zzylol]

Stack: main → this PR → series interning → no-forward mode. This PR merges the old #830 (per-sample ingestion) and #832 (columnar engine). It is built directly on main; #829 (concurrency) is deferred.

Why

Each pane build buffered every raw sample, turned each one into an executor Sample with an owned copy of its label map, then into a row of Values, and built the summary state from those rows. Query-time raw-sample windows and QueryRun builds took the same path. ASAPPlanner main (pinned ecea897f) runs its operators on Arrow batches and exposes StateColumnUpdater (the per-sample summary state update the batch SummaryAgg runs) and Batch::try_from_columns, so a runtime can build summary states per sample and install them as a stored pane without rows.

What

• Per-sample ingestion. A stored output whose build is Scan → TimeShift* → TimeRange(pane) → SummaryAgg, with per-series groups (the group columns include the series identity), is built per sample:
  • Per series, once: the Scan's label matchers (the executor's RawSampleSelector) and the series' text cells, cached.
  • Per sample: StateColumnUpdater::update on the open pane's summary state for that series, in arrival order. Stale markers are skipped, as the selector drops them.
  • At pane close: the states are sealed (StateColumnUpdater::seal), and the pane is one batch built with Batch::try_from_columns (group texts, the pane's timestamp, the state column), stored in the SDS.
  • Other shapes keep the plan's own build over the pane's buffered samples: a row filter, a keyed or shared-grid summary family, operators between the TimeRange and the SummaryAgg, matchers on value or time, or groups that are not series.
• Columnar raw-sample windows (series.rs). A RawSampleWindow's input batch is built column by column from the samples' times and values and each series' cached cells, with the selector's semantics: window bounds, latest sample per series for an instant selector, stale markers never rows. Used for query-time windows, QueryRun builds and the fallback build. No Sample and no row is made per sample.
• Rows only at the answer. The SDS stores a Pane (the build's batches), bound to a query root through Operator::source without copying. Batch::to_rows is called only in result_samples, which encodes the Prometheus JSON answer. Batch::try_new (rows to a batch) is no longer used.
• Compile once. The plan is parsed and compiled once, on the installing thread, and moved to the engine thread (it is Send now). With one engine thread there is nothing to share through Arc yet; feat(data-plane): sharded ingestion workers, a query thread pool and a shared SDS #829 does that when it lands.

How

• PerSampleSpec (one per definition) reads the SummaryAgg payload for the value, time and group columns, as the executor's lowering does. It finds the output timestamp's offset from the pane end by running the plan's build once over one row.
• OpenPanes numbers the series and holds one StateColumnUpdater per open pane. A pane's rows are the series that took a sample, in group-key order, as the batch build emits them.
• A summary state update error (e.g. a counter sample not after the series' previous one) fails that pane, which is counted in panes_failed, as a failed batch build was before.

Usage

No new flags. With the examples' plans without raw data, the data plane logs pane build per_sample=true for each ingestion-time stored output at startup, and /api/v1/status/asap counts are unchanged in meaning.

Before this PR

Per pane: one Sample (owned label map) and one row per raw sample, grouped into states by the executor's row path, and the result converted from rows back into a batch with Batch::try_new. Query-time raw-sample windows took the same Sample path.

After this PR

Per raw sample: one cached-series lookup and one in-place summary state update. Per pane: one try_from_columns. Rows are read only to encode the answer.

Verification

cargo test --workspace --locked passes (fmt and clippy not run for this draft; the benefit harness was not rerun). Existing tests are unchanged except that the SDS holds Pane (SummaryStore<Pane> in query_run.rs). New:

• per_sample::per_sample_panes_equal_the_batch_build: for Example 1 (counter rate and sum, 10 s panes), Example 3b (KLL, 1 min) and Example 4b (KLL, 10 min) without raw data, every pane built per sample equals the plan's own build of the same samples (Precompute::batch), row by row, with the same counters. Exact summary states are compared by serialized bytes, KLL by its quantiles at every percent (KLL seeds its compaction randomly, so two batch builds are not byte-identical either). Every ingestion-time stored output of those plans runs per sample.
• per_sample::per_sample_panes_apply_label_matchers.
• Answers still equal the Planner's query_time_reference (query_run::*, query::*).