feat(types): family-capability table and executor tumbling pane merges

[zzylol]

Supersedes #592 and #591, which stay open for now. Part of #509 (Example 3, tumbling panes).

Why

In #509 Example 3, a window is built as tumbling panes: one SummaryAgg per pane, combined by a SummaryMerge. Before the planner can emit that form, three things must hold:

• Planner rules and the executor need one place that says, for each family:

  • whether it merges;
  • how the merged inputs' selections must relate;
  • how each readout's error after a merge relates to its error from one build.

  Design decision: a generic family-capability table, with the IR left permissive.

• The executor must run a pane plan.

• Execution timing must accept SummaryMerge.

What

• Family-capability table. types::post_asap::family_capability adds FamilyCapability::of(&FieldDataType), which returns:
  • merge: Option<MergeCapability>:
    • inputs: SelectionRelation::{Disjoint, DisjointInTime, OverlapAllowed}
    • error: MergedError::{Exact, SameAsUnionBuild, Approximate}
    • heap_readouts: the merged error of readouts that read heavy-hitter heaps (distinct count, entropy, top-k), when it differs from error
  • subtractable
• FamilyCapability::merged_error(&SketchStatistic) gives the merged error per readout.
• Approximate means the merge is allowed but the merged readout is evidence only, for Stage 3 to judge (docs: propose workload-wide planning, summary sharing, and materialization #509 planner-layering.md, SummaryMerge).

Family	Merge inputs	Merged error	Subtracts
exact Sum, Count	disjoint	exact	yes
exact Min, Max	overlap allowed	exact	no
exact Rate, Increase	disjoint in time	exact	no
KLL, DDSketch	disjoint	same as one build over the union	no
HLL, KMV	overlap allowed	same as one build over the union	no
CMS, CountSketch	disjoint	same as one build over the union	yes
CMS/CountSketch with heap	disjoint	point counts: same as one build over the union; top-k: approximate	no
UnivMon	disjoint	approximate (every readout, at the current asap_sketchlib pin)	no
Hydra over CMS or CountSketch	disjoint	same as one build over the union	yes
exact IRate, Hydra over KLL, Theta, non-summary types	no merge		no

• IR.
  • SummaryMerge stays structurally valid for any family; the IR does not reject a merge by family.
  • Coverage applies each family's selection relation.
  • Execution timing implements SummaryMerge.
• Executor.
  • Runs tumbling pane merges, from executor: support tumbling pane merges (per-entity TimeShift panes, temporal merge timestamp) #591.
  • Refuses a merge, at query time and in precompute, when the table gives the family no merge.
  • Orders counter (Rate, Increase) pane states by time before merging them, so any input order gives the single-build result.
  • A SummaryMerge with no inputs is an invalid DAG, not a panic.

How

• Selection relations (from types: evaluation-relative coverage time and mergeable summary families #592). SummaryCoverage::derive consults the table:

  • OverlapAllowed skips the disjointness check, and the union of identical selection boxes is that box.
  • DisjointInTime requires the inputs' time windows not to overlap (CoverageError::NotDisjointInTime). Counter states join consecutive windows in time order (IncreaseAccumulator::merge_pair, with reset handling), so states disjoint only by a label would interleave in time.
  • types: evaluation-relative coverage time and mergeable summary families #592's evaluation-relative coverage time is not ported. Main's derived coverage already reads it from TimeRange over TimeShift.

• Heaps. A heavy-hitter heap keeps only the heavy items of its own part, so an item that is heavy over the window but in no single pane is lost.

• UnivMon at the pin. At the pinned asap_sketchlib (5f03ccb), UnivMon::merge rebuilds each layer's heap from the parts' heap keys, and L2 and distinct count go through that heap. A review measurement (heap 8, 20 panes) found:

  • L2 error of 6.8% for a single build, against 23.8% after the merge;
  • the heavy key in 0 of 100 merged heaps.

  A TODO(univmon-certification) marks the change for when the pin moves past upstream feat: rewrite uniqExact to COUNT(DISTINCT ...) #221: L2 becomes same as one build over the union, distinct count is revisited, and entropy stays approximate.

• KMV. asap_sketchlib's KMV (src/sketches/kmv.rs at 5f03ccb and 932ab4a) hashes every item with CANONICAL_HASH_SEED. KMV::merge keeps the k smallest distinct hashes, which is the state of the union, and it ignores repeated hashes.

• Timing. SummaryMerge inputs must be summary state at either phase. A merge at ingestion time needs every input at ingestion time. This follows refactor(planner): replace the summary lifecycle with #509 materialization options #667's Materialization assignment, so apply_materialization_timings covers both cases:

  • stored panes merged at query time (Pattern B1);
  • panes rebuilt at query time (Pattern B2).

• Executor.

  • A per-entity SummaryAgg looks through one TimeShift to its Scan.
  • A temporal SummaryMerge lowers to union → merge by series → scope timestamp, the same pattern as a per-series SummaryAgg.
  • require_sound_merge is shared by the query-time and precompute lowerings.
  • For a family whose inputs must be disjoint in time, SummaryMerge sorts its input states by their earliest sample time (ExactAccumulator::counter_window_start) before folding them with merge_pair.

Key code interfaces added / modified

crates/types/src/post_asap/family_capability.rs (new), mod.rs

/// What one summary family can do; planner rules and the executor consult it
/// to decide whether a composition over the family is sound.
pub struct FamilyCapability {                 // new: one per-family table of merge and subtract capabilities
    /// How the family's states merge; `None` when the family has no sound merge.
    pub merge: Option<MergeCapability>,
    /// Whether a state subtracts out of another of the same family
    /// (reserved for `SummarySubtract`).
    pub subtractable: bool,
}
/// How a family's states merge.
pub struct MergeCapability {                  // new: what a merge requires of its inputs and what error it yields
    /// How the selections of the merged states must relate.
    pub inputs: SelectionRelation,
    /// How the merged state's error relates to its parts' errors.
    pub error: MergedError,
    /// The merged error of readouts that read heavy-hitter heaps, when it
    /// differs from `error`; `None` otherwise.
    pub heap_readouts: Option<MergedError>,   // cardinality, entropy, top-k readouts
}
/// How a merge's input selections must relate to keep the family's guarantee:
/// no shared rows, non-overlapping time windows (counters), or overlap allowed (idempotent).
pub enum SelectionRelation { Disjoint, DisjointInTime, OverlapAllowed } // new
/// How a merged state's error relates to its parts: exact, the guarantee of
/// one build over the union, or approximate (no guarantee; evidence only).
pub enum MergedError { Exact, SameAsUnionBuild, Approximate }          // new

impl FamilyCapability {
    /// The capability of `family`; families not listed have no capability
    /// and fail closed.
    pub fn of(family: &FieldDataType) -> Self;                                  // new: the table lookup
    /// The merged error of `readout` from a merged state of this family
    /// (heap readouts use `heap_readouts`); `None` when there is no sound merge.
    pub fn merged_error(&self, readout: &SketchStatistic) -> Option<MergedError>; // new
}

// post_asap/mod.rs
/// The family-capability table module.
pub mod family_capability; // new
/// Re-exports the table's types at `post_asap::`.
pub use family_capability::{FamilyCapability, MergeCapability, MergedError, SelectionRelation};

crates/types/src/ir/summary_coverage.rs

/// Why coverage could not be derived for a summary node (type stays; one variant added).
pub enum CoverageError {
    /// The node, or a merge input, is not a `SummaryAgg` or `SummaryMerge` (internal bug in the caller).
    NotSummary,          // unchanged
    /// A `SummaryMerge` with no inputs (invalid DAG).
    EmptyMerge,          // unchanged
    /// A merge's inputs compute different things (different family, input or grouping); unsound as written.
    DefinitionMismatch,  // unchanged
    /// A merge's inputs are not proven disjoint and the family counts shared
    /// rows twice; the merge is unsound as written.
    PossibleOverlap,    // now only for families that count shared rows twice
    /// A counter-state (Rate/Increase) merge's inputs are not proven disjoint
    /// in time; the merge is unsound as written.
    NotDisjointInTime,  // new: counter-state merge whose inputs overlap in time
}

impl SummaryCoverage {
    /// Coverage of a `SummaryAgg` or `SummaryMerge`; a merge needs inputs with
    /// one definition whose selections relate as the family's table requires.
    // a merge now checks inputs against FamilyCapability::of(family).merge.inputs
    pub fn derive(node: &OperatorNode) -> Result<Self, CoverageError>;
}

crates/types/src/ir/timing.rs, post_asap/execution_data_state.rs

// validate_asap: ASAPOp::SummaryMerge { children } now has a timing contract
// (was ExecutionDataStateError::UnimplementedOperator): children must be summary
// state; an ingestion-time merge needs ingestion-time children, else
/// Produced when a child's data state is not accepted on that edge: a
/// malformed DAG (bad input).
ExecutionDataStateError::IllegalChildDataState { edge: "SummaryMerge.children", child: s }

crates/asap-physical-operators/src/physical_planner/mod.rs, precompute.rs

// new: shared by the query-time and precompute lowerings; rejects families without a sound merge
/// Errors unless the family-capability table gives the merged state's family
/// a sound merge; lowering calls it before running a `SummaryMerge`.
pub(crate) fn require_sound_merge(input: &SchemaRef) -> Result<(), Error>;

crates/asap-physical-operators/src/summary_kernels/exact.rs, operators/summary/mod.rs

impl ExactAccumulator {
    /// The earliest sample time of a counter state over all its populations;
    /// `None` for an empty or non-counter state. Used to order panes before a merge.
    pub fn counter_window_start(&self) -> Option<i64>; // new: earliest counter sample time
}
// merge_summary: orders DisjointInTime (counter) states by counter_window_start before folding

Usage, end to end (a 5-minute query answered from five 1-minute panes; numbers from a scratch run of tests/tumbling_pane_merge.rs's harness at 3a25450d). The planner's window-composition rule that emits panes comes later, so step 2 rewrites the selected plan into panes by hand, as that test does:

// 1. What the user writes: per-series p99 over the last 5 minutes, within ε = 0.05.
//    `single` is the plan the planner selects for it: one KLL { k: 52 } SummaryAgg per series
//    over a 5-minute TimeRange.
let single = single_build("quantile_over_time(0.99, m[5m])", AccuracyTarget::Epsilon(0.05));
// `family` is that SummaryAgg's family, the key step 3 looks up in the table.
let family = /* Sketch(Kll { k: 52, .. }, PerSubpopulationInstance), from single's SummaryAgg */;

// 2. The pane form. `panes` is the same query with the SummaryAgg rebuilt as five panes
//    (TimeRange(1m) over TimeShift(i·1m), i = 0..4) under one SummaryMerge; `pane_builds`
//    holds those five SummaryAgg nodes, which step 4 stores at ingestion time.
let (panes, pane_builds) = tumbling_panes(&single);

// 3. The family-capability table, consulted by coverage (SummaryCoverage::derive) and by the
//    executor's require_sound_merge before a SummaryMerge binds. `capability` holds KLL's row:
//    merge { inputs: Disjoint, error: SameAsUnionBuild, heap_readouts: None }, subtractable: false.
//    So the five disjoint panes may merge, and the merged p99 has one 5-minute build's error,
//    which is why the panes keep the single build's guarantee.
let capability = FamilyCapability::of(&family);
capability.merged_error(&SketchStatistic::Quantile { q: 0.99 });   // Some(SameAsUnionBuild)

// 4. Execution, over 2 series x 4 samples per minute (series a: 1..20, series b: 101..120).
//    `expected` holds the single build's rows; `rebuilt` the panes rebuilt at query time;
//    `retained` the panes built in five ingestion runs (MatAtIngestion), then merged at query time.
//    apply_materialization_timings now accepts the SummaryMerge in both assignments.
let expected = execute(&single);
let rebuilt = execute(&panes);
let retained = execute_retained(&panes, &pane_builds);
// What the user gets, all three identical, timestamped at the evaluation time 300,000 ms:
//   {series="a"} 20.0    {series="b"} 120.0

// 5. A counter: increase(m[5m]), exact. Its table row is
//    merge { inputs: DisjointInTime, error: Exact }, so the executor's SummaryMerge sorts the pane
//    states by ExactAccumulator::counter_window_start before merge_pair joins them across the
//    pane boundaries. Single build, rebuilt and retained panes all return
//   {series="a"} 20.0    {series="b"} 20.0

Before this PR

• apply_materialization_timings returned UnimplementedOperator { "SummaryMerge" }, so no pane plan could be exported.
• The executor rejected a per-entity pane build over TimeShift with "per-entity summary requires a resolved source".
• Nothing recorded whether a family's merge is sound, or what a merged readout's error is.
• Overlapping HLL or Min/Max inputs failed coverage with PossibleOverlap.

After this PR

A pane plan (5 one-minute panes under a SummaryMerge) exports and executes for sum_over_time(m[5m]), increase(m[5m]) and quantile_over_time(0.99, m[5m]). It returns the same per-series rows as the single 5-minute build, timestamped at the evaluation time, in two ways:

• panes rebuilt at query time;
• panes built in ingestion runs, then merged at query time.

SummaryMerge(HLL us, HLL us)                 -> coverage {region ∈ {us}}
SummaryMerge(KLL us, KLL us)                 -> PossibleOverlap
SummaryMerge(Increase api, Increase web)     -> NotDisjointInTime (same window)
SummaryMerge(Increase [-1m,0], [-2m,-1m])    -> coverage (-2m, 0]
merged_error(UnivMon, L2)                    -> Approximate (current pin)
merged_error(CmsWithHeap, TopK)              -> Approximate
merged_error(CmsWithHeap, PointCount)        -> SameAsUnionBuild

Verification

Each new test was written first and failed before its change.

• crates/types/tests/summary_merge_structure.rs:
  • the table, including the Hydra, KMV, heap and counter rows;
  • the merged error per readout;
  • the IR accepts a merge of IRate states;
  • HLL and exact Max merge overlapping states, while Sum and KLL do not;
  • SummaryMerge timing under the query-time, B1 and ingestion-time assignments.
• crates/types/tests/summary_coverage.rs: Increase panes merge when disjoint in time, but not when disjoint only by a label.
• crates/integration-tests/tests/tumbling_pane_merge.rs: exact Sum, exact Increase and quantile pane plans match the single build on both paths.
• crates/asap-physical-operators/tests/:
  • physical_semantics.rs: 60 compacted KLL panes (k = 200, 1000 values each) merge within the calibrated 99% rank error, 2.296/k^0.9723.
  • precompute_population.rs: a precompute merge of IRate state is rejected, and one of Sum state compiles.
  • summary_projection.rs: a SummaryMerge without inputs is rejected.
  • physical_semantics.rs: Increase panes A, B, C merged in the order A, C, B, with a counter reset inside B, give the single build's increase.
• An executor test of a pane plan whose family has no merge was dropped. No family the executor can build lacks a merge: IRate and shared Hydra grouping have no executor kernel. So the table check is covered through the precompute lowering (precompute_population.rs).
• Run on head 5b5fc9f6, rebased on main 80fc72e1:
  • cargo fmt --all -- --check: passed.
  • cargo clippy --workspace --all-targets --all-features --locked -- -D warnings: passed.
  • cargo nextest run --workspace: 1663 passed, 46 skipped.
• Screenshots and measurements: not applicable. The UnivMon numbers above come from the review, not from this PR's tests.

Limitations and follow-ups

• There is no planner-side pane rule, pane-width choice or compact pane-chain node yet. The pane plans in the tests are written by hand. The planner side comes in the follow-up.
• The table holds only mergeability facts. The summary-capability facts (which readouts a family answers) are added as fields of FamilyCapability later.
• UnivMon readouts are approximate after a merge until the pin moves past upstream asap_sketchlib feat: rewrite uniqExact to COUNT(DISTINCT ...) #221 (see the TODO).
• The executor has no Hydra merge kernel yet. The table marks Hydra over CMS or CountSketch mergeable for when the kernel lands.
• The executor has no KMV kernel (Executor has no KMV kernel although Pass 1 offers KMV for distinct counts #697), although Pass 1 offers KMV for distinct counts.

[zzylol]

Interface and efficiency review of this PR. These items land in the Pass 2 *Candidate rename PR after #717, alongside the #707 items (Readout → StatisticKind, subtractable removed).

Interfaces

• One table per family. Today two separate matches (merge_facts, readouts) feed MergeCapability { inputs, error, heap_readouts: Option<MergedError> }, and merged_error then special-cases entropy and top-k. This becomes
  FamilyCapability { merge_inputs: Option<MergeInputRule>, readouts: &'static [(StatisticKind, MergedError)] }.
  Each readout states its own merged error, and MergeCapability and heap_readouts go away. answers(s) and merged_error(s) are both lookups in that one list.
• One overlap error. CoverageError::PossibleOverlap and NotDisjointInTime become InputsOverlap { required: MergeInputRule }. A PaneChain whose whole-window expansion fails gets its own variant; it currently reports the misleading NotSummary.
• Plainer names.
  • SelectionRelation becomes MergeInputRule { Disjoint, DisjointInTime, MayOverlap }.
  • MergedError::SameAsUnionBuild becomes AsIfBuiltOverUnion.
  • ExactAccumulator::counter_window_start becomes earliest_sample_ms.

Efficiency

• SummaryCoverage::derive checks a merge's inputs pairwise, O(n²·s²) for n inputs with s selection boxes each. A day of 1-minute counter panes (n = 1,440) is about 1M box comparisons. Because coverage() is not cached, validation repeats this for every node that reads the merge. The planned fix sorts inputs by window start and compares neighbours, O(n log n), for DisjointInTime. For Disjoint, the same time order prunes pairs whose windows cannot overlap. Validation will also stop deriving coverage again for every node.

Minor: this PR body's interface comment says heap readouts cover "cardinality, entropy, top-k". The code (merged_error) applies the heap error only to entropy and top-k; a distinct count takes the state's error. The one-table change makes this explicit per readout.