refactor: keep storage codecs, drop duplicated summary kernels

[zzylol]

Stacked on #807.

Why

#793 copied Planner's old summary kernels into crates/asap_summary_state so the store could keep their byte formats. That left two implementations of every kernel (Sum/Min/Max/keyed exact states, KLL, DDSketch, HLL, CMS/CountSketch with and without heaps, UnivMon, Hydra, the updater factory) plus physical.rs converting between them wherever stored state entered or left a Planner DAG. The approved layering is that Planner owns computation and the backend owns storage formats, decoding and readout binding.

What

• The store holds Planner kernel states (asap_physical_operators::AggregateCore) directly.
• stored_state::codec is the one backend view of those states. The StoredState extension trait gives each one a persisted tag, an AggregationType and a byte encoding. decode(tag, bytes) returns a Planner kernel. empty_like replaces the old window-reset path. check_storable guards the store's append boundary, so a state without a codec is refused instead of being written as empty bytes.
• stored_state::decoders holds the edge wire formats (protobuf envelopes, msgpack frames, delta frames, the Sum payload) as functions over sketchlib types. The ingest and read paths now share these functions. Before, CMS, CountSketch and heap frames each had two decoders.
• Deleted: summary_kernels/* (23 files), physical.rs, the backend updater factory, the legacy query_statistic/aux_stats/SingleSubpopulationAggregate/MultipleSubpopulationAggregate surface, and asap_types::traits. Tests read through Planner's estimate/readout. SummaryState now wraps Planner kernels and merges with Planner's merge_with. Exact readouts delegate to Planner's readout::exact_readout.

Before this PR

A DDSketch frame decoded into the backend's DDSketchAccumulator. On the way into a Planner DAG it went through physical::to_physical, which checked sample_p and copied inner into Planner's DDSketchAccumulator. Outputs came back through from_physical.

After this PR

The same frame decodes straight into Planner's DDSketchAccumulator, and the store keeps that value. DAG inputs are Arc::clones of it. check_storable rejects states that have no codec, the case where from_physical used to fail.

Lines: 68 files, +2,651 / −13,932 (asap_summary_state alone: +1,911 / −12,809). New golden, decoder and rollup tests are included in the added lines.

Compatibility

These current formats round-trip unchanged. Each family is tested against golden bytes written by the pre-change code, checking encode(decode(x)) == x and the readouts:

• sketchlib msgpack for DDSketch, HLL, KLL, CMS, CountSketch, CMS/CountSketch heaps and Hydra
• PlannerExactAccumulatorV1
• WeightedFrequencyV1
• UnivMonAccumulator
• NativePhysicalOutputV1 batches

Edge wire formats are unchanged.

Dropped, as the owner approved for this development stage. Each now fails with a named, versioned rejection instead of being misread:

• Durable exact tags SumAccumulator, IncreaseAccumulator, MinAccumulator, MaxAccumulator, KeyedSumCountAccumulator, KeyedCounterState (and KeyedMinState/KeyedMaxState). decode returns "stored format X is retired". Immutable maintenance inputs fail with that reason. The best-effort durable query union logs it and skips the entry, the same way it already skips unreadable parts.
• Native batch codec SumAccumulatorV1.

Behavior changes:

• Edge-sampled frames are rejected: DDSketch, HLL, CMS and CountSketch envelopes with 0 < sample_p < 1 fail to decode. Planner kernels have no place to carry sample_p. Before, the stored-bytes readout silently ignored it, and only the legacy in-memory readout rescaled counts.
• OTLP SketchEnvelope attribute payloads decode into the matching Planner kernel. Before, they were an opaque accumulator whose merge kept the left side.
• Unkeyed Planner exact MIN/MAX panes feed the MIN/MAX rollup again. Its only producers had been the legacy Min/MaxAccumulator, which exact states stopped producing. A pane that does not extend the series contiguously (a late correction, a replacement or a gap) now permanently invalidates that series' rollup, so readers fall back to the exact panes instead of serving an uncorrected extremum. Regression tests planner_exact_min_panes_feed_the_min_rollup and min_rollup_stops_answering_after_a_late_correction fail on the previous logic.
• A DDSketch envelope arriving on the proto-delta channel that fails to decode (for example, a sampled one) is rejected. Before, it fell through to the bucket-delta decoder and could apply as an empty delta. Regression test: sampled_ddsketch_envelope_on_delta_channel_is_rejected.
• DDSketch PROTO ingest uses the same decoder as the read path (DdSketch::from_proto), so negative and zero buckets are kept. Before, ingest used from_raw, which kept positive buckets only.
• CMS read-path frames get the ingest decoder's dimension checks.

Remaining

Two shims are kept, marked in the code:

• asap_summary_state::univmon, because Planner's UnivMon has no public sketch or byte form.
• The msgpack round trips in frequency_kernel/decode_exact.

A small Planner PR would let both go by adding:

1. UnivMonAccumulator::from_sketch(UnivMon) -> Result<Self> and UnivMonAccumulator::sketch(&self) -> &UnivMon (or a byte codec). This deletes the UnivMon shim.
2. Public WeightedFrequency::algorithm(), shape() and kernel byte access (to_bytes/from_bytes). This removes the msgpack round trip in frequency_kernel/frequency_state.
3. A validating ExactAccumulator decode (or a public constructor from populations). This removes the serde mirror in decode_exact.
4. AggregateCore::as_any_mut (or into_any). This lets ingest delta frames mutate a cached base in place instead of clone-and-replace, and lets SummaryState::merge_same_family avoid a second copy after Planner's merge_with.
5. Optional edge sample_p on the DDSketch/HLL/CMS kernels (or a scaled readout), to re-admit sampled edge frames.

Validation

• cargo fmt --all -- --check
• cargo clippy --workspace --all-targets --locked -- -D warnings
• cargo test --workspace --locked --lib
• cargo test -p control_plane --locked --tests
• cargo test -p data_plane --locked --test asapquery_compatibility_process_e2e -- --test-threads=1
• Independent review by a separate agent; findings addressed.

🤖 Generated with Claude Code