feat(data_plane): implement SummaryExecutor for cumulative quantile/cardinality (Step C, first cut)

[zzylol]

Summary

Step C of the plan-shaped-serving migration (data_plane/docs/l4node-plan-executor-design.md, #409). Stacked on #410 (rev pin bump) since asap_sketch::exec::SummaryExecutor doesn't exist before that.

Scoped to quantile/cardinality queries (DDSketch/Kll/Hll) and the Frequency family's bare total (CMS/CountSketch/CMS-with-heap/CountSketch-with-heap — count/sum with no specific item key), both cumulative (instant) and per-window (matrix/range) — see the module doc in summary_executor.rs for exactly what's in/out of scope. Explicitly deferred, erroring rather than silently mishandled: SketchQuery::TopK (its answer shape — K items per timestamp — doesn't fit this executor's Value = Vec<(i64, f64)> at all; needs a Value type redesign, a separate decision) and PointCount with a named item key (the filter value isn't carried by SketchQuery or available in readout's signature).

Seven commits:

1. refactor(delta_apply): generalize cumulative_hll_state to all sketch families — the DD/KLL/HLL cross-sid merge primitive.

2. feat(data_plane): implement SummaryExecutor for cumulative quantile/cardinality — QueryExecutionContext implements the trait; exact (SummaryKind, SummaryParams) matching; lazy fetch_state/merge_states, real merge in readout.

3. feat(data_plane): per-window (matrix/range-query) readout for SummaryExecutor — per_window_rolling_states, cross-sid per-window merge.

4. fix(data_plane): address PR review feedback on SummaryExecutor — removed the now-redundant cumulative_hll_state wrapper, rewrote comments to describe current design instead of history, fixed a real inefficiency (double query_range fetch per candidate — SidHandle now carries the already-fetched series via Rc).

5. feat(data_plane): unify Frequency family into SummaryState — CMS/CountSketch/CMS-with-heap/CountSketch-with-heap folded into the same SummaryState/merge_same_family pipeline DD/KLL/HLL already use (not a separate parallel implementation — see the PR comment on this commit for why the original separate-file approach was wrong).

6. fix(data_plane): wire real asap_sketchlib::CountSketchWithHeap into SummaryState — CmsWithHeap and CountSketchWithHeap were still sharing one underlying wire type (CountMinSketchWithHeap, CMS's min-over-rows estimator) because asap_sketchlib had no separate CountSketchWithHeap (median-of-signed-rows estimator) type. Rather than work around that gap in data_plane, added the real type upstream (ProjectASAP/asap_sketchlib#78, now merged) and wired it in here: new decode_cs_with_heap_from_msgpack{,_delta} decoders, SummaryState::CountSketchWithHeap now holds the genuine type. The two variants holding distinct Rust types means a cross-family merge is now a type-level impossibility, not just a runtime check.

7. fix(sketch_reducer): route CountSketchWithHeap through its own estimator — the same conflation bug, found separately: sketch_reducer.rs's legacy decode_frequency_estimate (a different, still-live code path from SummaryState/delta_apply.rs) always decoded heap-bearing sids through CountMinSketchWithHeap and rebuilt a plain CountMinSketch regardless of the sid's real kind — silently wrong for CountSketchWithHeap's per-key topk/FrequencyEstimate queries. Split the collapsed match arms in decode_frequency_total, decode_frequency_estimate, and the FrequencyTopk per-frame heap-decode loop to dispatch on the sid's actual kind.

Also adds asap-sketch/asap-ir as direct data_plane dependencies (pin-matched to control_plane's — required even though control_plane::asap_tier_implement already returns Vec<Rc<L4Node>>, since implementing a trait on / matching a type's variants requires importing its defining crate directly).

Leaving this unmerged for review, per the same reasoning as #410/#161: this changes the live query-serving path's dependency surface and introduces a new, not-yet-wired-in code path — worth a human look before anything merges, especially the scope boundary (what's covered vs. deferred) and the find_candidates/lazy-state design.

Test plan

• cargo test -p data_plane --lib summary_executor — 12 tests, all passing. The ones that matter most:
  • two_sids_same_group_actually_merge_not_just_first / two_cms_sids_same_group_totals_actually_merge — proves real cross-sid merge for both the Sketch and Frequency families.
  • two_sids_different_groups_produce_two_series_not_one_merged_blob — the feat(engine): read precomputes from SketchIndex when attached #159 fix, exercised end-to-end.
  • per_window_matrix_produces_multiple_points_for_one_sid / per_window_matrix_merges_across_sids_per_window — matrix output correctness.
  • topk_query_is_explicitly_unsupported_not_silently_wrong — the scope boundary errors instead of returning wrong data.
• cargo test -p data_plane --lib delta_apply — 11 tests, all passing, including cms_with_heap_and_count_sketch_with_heap_are_not_the_same_family (now backed by genuinely distinct types) and count_sketch_with_heap_full_and_delta_decode_via_new_asap_sketchlib_type (new — proves both the FULL and DELTA-HEAP decode paths reconstruct the same matrix/estimate() as the in-memory sketch they were encoded from).
• cargo test -p data_plane --lib frequency_heap_tests — new decode_frequency_estimate_uses_each_kinds_own_estimator, built via real update() calls on both kinds and asserting decode_frequency_estimate reproduces each kind's own in-memory estimate() truth.
• cargo test -p data_plane sketch_db:: — 294 passed, confirms the evaluate_cardinality_global refactor is behaviorally identical.
• cargo test -p data_plane --lib — full lib suite: 908 passed, 2 ignored. Only pre-existing, unrelated failures elsewhere in the workspace, verified identical on the parent commit before this branch's changes existed.
• cargo build --workspace — clean, against the real (now-merged) asap_sketchlib main — no local [patch] workaround needed.
• cargo clippy -p data_plane --lib -- -D warnings — zero warnings in the touched files.

🤖 Generated with Claude Code

[zzylol]

You don't need to keep it separately if cumulative rolling state func covers it.

[zzylol]

Removed — evaluate_cardinality_global now calls cumulative_rolling_state directly and reads out via RollingState::cardinality() instead. Confirmed via the existing 294-test sketch_db suite passing unchanged.

[zzylol]

Added the range-query (matrix) follow-up in a second commit — SummaryExecutor now covers both cumulative (instant) and per-window (matrix/range) quantile/cardinality queries, not just cumulative.

• delta_apply.rs: new per_window_rolling_states, generalizing per_window_evaluate the same way cumulative_rolling_state generalized cumulative_hll_state — returns each window's reconstructed state instead of an already-evaluated scalar, so callers can merge same-window states across sids before evaluating. per_window_evaluate now delegates to it (verified behaviorally identical: same 294-test sketch_db suite passes unchanged).
• summary_executor.rs: readout dispatches on is_cumulative to readout_cumulative (existing) or the new readout_per_window, which unions windows across a group's sids and merges same-window states (mirrors SummaryMerge's "fold whatever's present" semantics from chore(precompute_engine): delete unused RawPassthroughSink #161/ASAPController#159 — a sid missing a window doesn't drop the window, just doesn't contribute to it). Drops carry-in base windows ending before t0, matching evaluate_core's existing identical filter on the legacy path.

3 new tests (multi-window correctness, cross-sid per-window merge, carry-in-base filtering) — all passing, plus the existing 6 unaffected. Same pre-existing unrelated e2e failures as before, still unrelated (verified).

Remaining scoped-out items, unchanged from before: Frequency family (TopK/CMS) and ExactAgg intents (need caller-side handling, not readout).

🤖 Generated with Claude Code

[zzylol]

Could you make the code comments not showing step X, phase Y etc, and not talking about the detailed wrong case before, but just why currently we have this design and how it is.

[zzylol]

Rewrote the module doc and every inline comment to describe the current design and its rationale, not the sequence of steps/gaps/issue numbers that produced it (no more "Step C", "#409", "ASAPController#161", "gap #409 flagged", etc.).

[zzylol]

Can you check any inefficiency in SummaryExecutor implementation, e.g., memory copy, not needed deserialization/serialization, any potential caching, any redundant indexing/inefficient indexing? Any extra memory usage, computation overhead?

[zzylol]

Efficiency review, and one real fix landed (pushed):

Fixed — double fetch + double decode. find_candidates called query_range(sid, t0, t1) per candidate just to read label values for the group key (that's the only place per-sid label values live). readout then called query_range again on the same (sid, t0, t1) to get the actual sample data. query_range returns an owned Vec<SketchTimeSeries> — building it clones every in-range sample's bytes — so this was doubling the clone/decode work per candidate sid, plus leaving a narrow window where the two calls could observe different data under a concurrent write between them. Fixed: SidHandle (the trait's Handle type) now carries the already-fetched series as Rc<SketchTimeSeries> from find_candidates through fetch_state/merge_states/readout, so query_range is called exactly once per candidate. Also folds delta_kind onto the handle at the same time (find_candidates already looks up the sid's SketchConfig for the exact-param-match check), removing a second with_instance lookup fetch_state used to make on its own.

Considered, not changed:

• readout's per-entry Vec<(i64, &SketchSampleState)> rebuild (flat_map().collect()) — a real allocation, but it's the exact same pattern sketch_reducer.rs's evaluate_core/evaluate_exact_agg already use throughout the existing codebase, not something new I introduced. Changing it to avoid the allocation would mean changing delta_apply.rs's function signatures to take an iterator instead of a slice — a bigger, more invasive change than this PR's scope.
• GroupKey: BTreeMap<String, String> label cloning per candidate — required by the trait's GroupKey: Clone bound and matches the BTreeMap<String,String> label-map convention already used throughout the reducer; no cheaper representation available without a broader label-interning change (which is what design-backend-plan-wire-format.md's RoutingIndex Tier-2 design already proposes for a different reason — group lookup, not just this).
• Rc clone cost elsewhere: fetch_state clones one SidHandle (an Rc bump + a Copy enum) per handle, merge_states just extends Vecs — no unnecessary deep copies anywhere in the group-assembly path.

Verified no regression: cargo test -p data_plane --lib summary_executor (9/9) and cargo test -p data_plane sketch_db:: (294/294) both still pass after the fix.

🤖 Generated with Claude Code

[zzylol]

Frequency family (CMS/CountSketch) follow-up landed, in two parts:

Part 1 (superseded by part 2, kept as commit history): started as a separate `frequency_apply.rs`, justified by "no rolling reconstruction needed for this family." That framing was wrong — re-examined the actual delta-decode implementations and found the Frequency family's "decode independently, then merge" is the same fold operation `delta_apply.rs` already uses for every DD/KLL/HLL delta case except two genuine sparse-in-place special cases (DD's bucket-index proto delta, HLL's register proto delta) — both of which also decode-and-merge for their other encodings.

Part 2: unified. `RollingState` → `SummaryState`, extended with `Cms`/`CountSketch`/`Heap` variants (Heap covers both `CmsWithHeap` and `CountSketchWithHeap` — they already share one wire representation and read out identically in the existing reducer). One `merge_same_family`/`cumulative_rolling_state`/`per_window_rolling_states` pipeline for all six sketch kinds now, not two parallel ones. `frequency_apply.rs` deleted. `summary_executor.rs`'s `CandidateKind` wrapper enum also dropped — no longer needed once there's one unified kind.

Real finding worth flagging explicitly: `SketchQuery::TopK`'s answer (K items per timestamp) doesn't fit this executor's `Value = Vec<(i64, f64)>` shape at all — forcing it in (e.g. returning only the top item) would silently drop data rather than error. Scoped out, with a test (`topk_query_is_explicitly_unsupported_not_silently_wrong`) proving it errors instead of returning something wrong. Needs a `Value` type redesign — a separate decision, not bundled into this PR.

Also refactored `evaluate_cardinality_global` to use `SummaryState` directly and dropped the now-redundant `cumulative_hll_state` wrapper this obsoletes (following up on the earlier review comment about it).

Actual scope covered now: quantile/cardinality (unchanged) + the Frequency family's bare total (no specific item key), both cumulative and per-window. PointCount with a named item key and TopK both explicitly unsupported, both with a documented reason in the module doc.

Test plan

• 12 tests in summary_executor (9 existing + 3 new), all passing — CMS total readout, CMS cross-sid merge (matrix addition, same property already proven for DD/KLL), and the TopK-errors-don't-silently-return-wrong-data test.
• sketch_db:: suite (294 tests) confirms the evaluate_cardinality_global refactor is behaviorally identical.
• Full suite: 905 passed, same pre-existing unrelated failures already documented on this PR.
• clippy -p data_plane --lib -- -D warnings: clean.

🤖 Generated with Claude Code

[zzylol]

Addressed both new comments (couldn't reply inline — the pending-review comments can't take a threaded reply while that review is still open, so posting here instead):

"there should be a CountSketchWithHeap here" (delta_apply.rs) — confirmed asap_sketchlib has no separate CountSketchWithHeap type at all (only CountMinSketchWithHeap), so I'd originally collapsed both CmsWithHeap and CountSketchWithHeap into one shared Heap variant. That was a real bug, not just a naming choice: a CMS-substrate heap and a CountSketch-substrate heap are different sketch algorithms that merely share a storage shape — merge_same_family's single Heap-Heap arm would have silently allowed merging one into the other (mathematically invalid, but type-checks fine since both wrap CountMinSketchWithHeap). Fixed: split into two distinct enum variants (CmsWithHeap/CountSketchWithHeap), still both backed by CountMinSketchWithHeap (no other type exists), but now merge_same_family's per-variant match rejects the cross-family case the same way it already rejects e.g. merging a Cms into a Kll. Added cms_with_heap_and_count_sketch_with_heap_are_not_the_same_family proving the fix.

"cumulative_rolling_state should be renamed to cumulative_summary_state to match SummaryState" — done, and per_window_rolling_states → per_window_summary_states too (same staleness, missed in the rename commit).

Pushed. 906/906 passing (main lib target), 12/12 summary_executor, 10/10 delta_apply (9 existing + the new regression test), clippy -D warnings clean.

🤖 Generated with Claude Code

[zzylol]

cumulative_rolling_state should be renamed to cumulative_summary_state to match SummaryState

[zzylol]

Follow-up on the "there should be a CountSketchWithHeap here" comment: rather than keep working around the gap in data_plane, added the real type to asap_sketchlib (ProjectASAP/asap_sketchlib#78, merged) and wired it in here in 61f3425. SummaryState::CountSketchWithHeap now holds the genuine asap_sketchlib::CountSketchWithHeap (median-of-signed-rows estimator) instead of aliasing the CMS-family CountMinSketchWithHeap (min-over-rows estimator). New decoders in decoders.rs, updated decode_full/apply_delta_bytes/total/topk_items in delta_apply.rs. The two SummaryState variants now hold genuinely different Rust types, so a cross-family merge is a compile-time impossibility, not just a runtime check — see the updated cms_with_heap_and_count_sketch_with_heap_are_not_the_same_family test plus the new count_sketch_with_heap_full_and_delta_decode_via_new_asap_sketchlib_type regression test.