feat(asap-physical-operators): execute HydraCms grouped counts and filtered aggregates

[zzylol]

Why

G7a is the executor side of grouped and filtered counts (part of #509 and #580). The planner side, which offers HydraCms in Pass 1, is G7b and lands after #690. Before Pass 1 can emit a shared Hydra grid or a filtered SummaryAgg, the executor has to run both. G8 also needs a guarantee to rely on: a filtered SummaryAgg never runs at ingestion time.

Supersedes #600 (HydraCms kernel) and #607 (filtered aggregates). Both stay open for reference.

What

• HydraCms grouped counts and weighted sums. A SummaryAgg with SharedMultiSubpopulation { kind: HydraCms } grouping now executes. It builds one shared grid for all groups and emits one row per group. That group's row answers a valued PointCount, which is an item's count, or its summed weight, within the group. Two of these states merge with SummaryMerge. The plan contract it binds:

  • family Sketch(Cms{width,depth}, …), where width and depth match the HydraParams;
  • one item column, a non-null Utf8, Int64 or Bool;
  • weight Constant(1.0) with UnitCount (a count), or a weight column (Float64 or Int64) with a NonNegative weight domain, e.g. bytes per item per group. Each row's weight must be a non-negative integer within i32, a NULL weight adds nothing, and an unknown-sign weight is rejected at binding;
  • Reduce(group keys).

  Error and cost. A weighted point count overestimates by at most the Count-Min bound over the stream's total weight (‖stream‖₁), not its row count, at both Hydra levels. The cost per row is unchanged: one insert_many with the row's weight.

• Filtered aggregates. Per-measure filters on Aggregate and a row filter on a SummaryAgg now execute with SQL FILTER (WHERE …) semantics:

  • only rows where the predicate is true update the measure;
  • a group with no such row is kept: a count reads 0, and SUM/AVG/MIN/MAX and quantiles read NULL.

• A filtered SummaryAgg runs only at query time. Both precompute compilation and the binding of any SummaryAgg timed IngestionTime reject it.

• A quantile over a relation is nullable when it can have no value. This covers a nullable summarized column and an ungrouped reduction, in addition to the filtered case. In both the Aggregate and the SummaryEstimate realizing it, an all-NULL group or a global quantile over no row now reads NULL instead of failing the query.

• Measure filters over PromQL series identity are rejected. PromQL emits none, and its outputs are non-nullable.

How

• summary_kernels/hydra_cms.rs wraps asap_sketchlib::Hydra with HydraCounter::CM cells. It has no local copy of sketchlib logic.
  • It uses a one-column schema, so a group's whole label tuple is one subpopulation and rows do not fan out.
  • The readout is Hydra's median estimator.
  • Weights go through Hydra's update(.., count: Option<i32>), which calls the cell's Count-Min insert_many. Because those weights are i32, a negative, fractional or non-finite row weight fails the build with a clear error.
  • Merging rejects shape mismatches, because sketchlib's CMS merge panics on them. It also caps total weight at the limit of sketchlib's i32 counters.
• A new SharedSummaryBuild operator builds the grid. HydraCmsGroup is the row state: an Arc to the grid plus the group it reads.
• factory::create_hydra_cms builds the grid. The capability checks cover the update, the native family, the evaluation and the state payload.
• Aggregate gains filters (with_measure_filters). The per-group, keyed and shared summary builds gain filter (with_row_filter). Both fields are omitted from serialization when absent, so existing plans are unchanged.
• AggregateCore::is_empty (KLL, DDSketch) lets a nullable quantile read NULL for an empty state. An empty KLL quantile is now an error, not 0.0.
• capability::validate_ingestion_time_summary is the single ingestion-time check.
• Shared grids are accounted and merged once. AggregateCore::shared_state lets a row report a state it shares with other rows (the Hydra grid). Batch::bytes and SummaryMerge reserve each shared grid once, not once per group. SummaryMerge merges each distinct pair of grids once, memoized on the pair of Arcs, and every group of the pair reads the one merged grid.
• Types (ir/asap.rs, ir/non_asap.rs): quantile nullability follows the rule of SUM over a relation. It is nullable for a filtered measure, a nullable input column, or a global reduction, and the same rule is seen through SummaryMerge.
• The unfiltered SummaryAgg lowering moved, unchanged, into physical_planner::summary_build, so the filter is applied in one place. The binding-time rejections of filtered aggregates, of filtered summary updates and of all shared grouping are deleted.

Key code interfaces added / modified

crates/asap-physical-operators/src/summary_kernels/hydra_cms.rs, mod.rs, factory.rs

/// Module holding the HydraCms kernel; its items are re-exported from `summary_kernels`.
pub mod hydra_cms;                           // new; pub use hydra_cms::*
/// One Hydra grid of Count-Min cells shared by all groups of a grouped `SummaryAgg`;
/// answers an item's count or weight sum within a group.
pub struct HydraCms {                        // new type
    /// The library grid: `shared_rows x shared_columns` cells, each a Count-Min.
    inner: Hydra,                            // new: asap_sketchlib::Hydra with HydraCounter::CM cells
    /// Each cell's Count-Min width; merges require it equal.
    width: usize,                            // new
    /// Each cell's Count-Min depth; merges require it equal.
    depth: usize,                            // new
    /// Total weight added; kept within `i32` so sketchlib's `i32` counters cannot overflow.
    total: i64,                              // new
}
impl HydraCms {
    /// Builds an empty grid with these dimensions; fails (bad input) if any is 0.
    pub fn new(shared_rows: usize, shared_columns: usize, width: usize, depth: usize) -> Result<Self, KernelError>;
    /// `(shared_rows, shared_columns, width, depth)`; merges require equal shapes.
    pub fn shape(&self) -> (usize, usize, usize, usize);
    /// Adds `weight` occurrences of `item` in `group` (1 for a row count); fails if
    /// the weight is negative or the total weight would exceed `i32`.
    pub fn count(&mut self, group: &str, item: &str, weight: i32) -> Result<(), KernelError>;
    /// Estimated count (or weight sum) of `item` within `group`: Hydra's median
    /// over rows of each cell's Count-Min estimate; never below the exact value.
    pub fn point(&self, group: &str, item: &str) -> Result<f64, KernelError>;
    /// Merges two grids built over disjoint inputs into the grid of their union;
    /// fails on a shape mismatch or an `i32` total overflow.
    pub fn merge(&self, other: &Self) -> Result<Self, KernelError>;
    /// Approximate footprint of the whole grid, used for memory reservations.
    pub fn approx_memory_bytes(&self) -> usize;
}

/// One group's row state: a `SummaryAgg` emits one per group, all pointing at the same grid.
pub struct HydraCmsGroup {                   // new: per-group row state reading one shared grid
    /// The grid shared by every group of the build.
    pub grid: Arc<HydraCms>,
    /// The group label tuple this row reads from the grid.
    pub group: String,
}
/// Answers a valued `PointCount` for its group; a group total is rejected (no group-mass query).
impl AggregateCore for HydraCmsGroup {       // new impl
    /// A copy of the row: same `Arc` grid, same group.
    fn clone_boxed_core(&self) -> Box<dyn AggregateCore>;   // new
    /// This row as `Any`, for downcasting to `HydraCmsGroup` (merge, state checks).
    fn as_any(&self) -> &dyn std::any::Any;                 // new
    /// Mutable `Any` view, for in-place downcasts.
    fn as_any_mut(&mut self) -> &mut dyn std::any::Any;     // new
    /// Merges the two grids (HydraCms::merge) for the same group; errors (bad plan) on another
    /// family or another group, or on a shape mismatch or `i32` overflow.
    fn merge_with(&self, other: &dyn AggregateCore) -> Result<Box<dyn AggregateCore>, KernelError>;  // new
    /// A valued `PointCount` reads that item's count or weight sum in the group (grid.point);
    /// anything else, including a group total, is an error (unsupported readout).
    fn estimate(&self, query: &SketchStatistic) -> Result<f64, KernelError>;   // new
    /// The row's own footprint (struct plus group string); the grid is counted by shared_state.
    fn approx_memory_bytes(&self) -> usize;                 // new
    /// (grid pointer, grid footprint), so the grid is reserved and merged once for all groups.
    fn shared_state(&self) -> Option<(usize, usize)>;       // new
    // is_empty: trait default (false)
}

/// Builds the one grid a `SharedMultiSubpopulation { kind: HydraCms }` family shares
/// across its groups; fails if the family is not a valid HydraCms shape.
pub fn create_hydra_cms(family: &SummaryFamilyType) -> Result<HydraCms, String>;   // new (factory.rs)

crates/asap-physical-operators/src/summary_kernels/traits.rs, datasketches_kll.rs, dd_sketch.rs

/// The per-row summary state every kernel implements: update, merge, estimate, size.
pub trait AggregateCore {                    // trait stays; two default methods added
    /// A boxed copy of the state.
    fn clone_boxed_core(&self) -> Box<dyn AggregateCore>;   // unchanged
    /// The state as `Any`, for downcasts to the concrete kernel.
    fn as_any(&self) -> &dyn std::any::Any;                 // unchanged
    /// Mutable downcast, so a deployment can apply deltas and merges in place.
    fn as_any_mut(&mut self) -> &mut dyn std::any::Any;     // unchanged
    /// Merges with a state of the same family and shape, leaving both inputs unchanged.
    fn merge_with(&self, other: &dyn AggregateCore) -> Result<Box<dyn AggregateCore>, KernelError>;   // unchanged
    /// Answers a sketch readout; the default rejects every readout.
    fn estimate(&self, query: &SketchStatistic) -> Result<f64, KernelError> { /* Err */ }  // unchanged
    /// Approximate footprint for memory reservations, excluding a shared state; default 4096.
    fn approx_memory_bytes(&self) -> usize { 4096 }         // unchanged (doc now excludes shared state)
    /// True when no sample reached the state, so a quantile has no answer and reads NULL;
    /// kernels that do not track it report false.
    fn is_empty(&self) -> bool { false }                       // new; KLL and DDSketch override: empty quantile reads NULL
    /// Identity and footprint of a state shared with other rows (one Hydra grid), so
    /// reservations and merges count it once; `None` when nothing is shared.
    fn shared_state(&self) -> Option<(usize, usize)> { None }  // new; Batch::bytes and SummaryMerge count a shared grid once
}

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

impl Operator {
    /// A `SummaryAgg` whose groups share one HydraCms; each row adds its `weight` column's
    /// value (1 without one) to its `item`, and a NULL weight adds nothing.
    pub fn shared_summary_build(                                // new: builds SharedSummaryBuild
        input: SchemaRef,
        family: SummaryFamilyType,
        item: usize,
        weight: Option<usize>,
        groups: Vec<usize>,
    ) -> Result<Self, Error>;
    /// Updates the summary only from rows where `filter` is true (a filtered `SummaryAgg`);
    /// every group still gets a state, so an unmatched group reads an empty summary.
    pub fn with_row_filter(mut self, filter: Expression) -> Result<Self, Error>;                  // new: per-group, keyed and shared builds
    /// Gives each `Aggregate` measure an optional row filter; an unmatched group is kept
    /// with the empty-input value (0 for counts, NULL for SUM/AVG/MIN/MAX).
    pub fn with_measure_filters(mut self, filters: Vec<Option<Expression>>) -> Result<Self, Error>;  // new: SQL FILTER (WHERE ...) on Aggregate
}

crates/asap-physical-operators/src/operators/mod.rs, operators/unchecked.rs (serialized operator format; existing plans serialize unchanged)

/// The serialized form of a physical operator.
enum Kind {   // type stays; SharedSummaryBuild added, filter fields added
    /// Grouped exact aggregation.
    Aggregate {
        /// Group key column indices.
        groups: Vec<usize>,
        /// One reduction per output measure.
        measures: Vec<Reduction>,
        /// Per-measure row filters (SQL `FILTER (WHERE …)`); empty when none.
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        filters: Vec<Option<Expression>>,              // new
    },
    /// A per-group `SummaryAgg`: one summary state per group.
    SummaryBuild {
        family: SummaryFamilyType, value: usize, time: Option<usize>, groups: Vec<usize>,
        /// Row filter; rows where it is not true update no state, and their group is kept.
        /// Absent (the default) means every row updates.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        filter: Option<Box<Expression>>,               // new
    },
    /// A keyed `SummaryAgg`: one state per group, keyed by item columns.
    KeyedSummaryBuild {
        family: SummaryFamilyType, value: usize, items: Vec<usize>, groups: Vec<usize>,
        /// Row filter; rows where it is not true update no state, and their group is kept.
        /// Absent (the default) means every row updates.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        filter: Option<Box<Expression>>,               // new
    },
    /// A `SummaryAgg` with one HydraCms shared by all groups, adding each row's item weight.
    SharedSummaryBuild {                               // new: one HydraCms shared by all groups
        family: SummaryFamilyType,
        /// The item column counted within each group.
        item: usize,
        /// The weight column; each row adds 1 when absent (the default).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        weight: Option<usize>,
        /// Group key column indices.
        groups: Vec<usize>,
        /// Row filter; rows where it is not true update nothing. Default none.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        filter: Option<Box<Expression>>,
    },
    /// Rows supplied by the caller (not serialized).
    Source(Vec<Batch>),                                                        // unchanged
    /// One constant value of `dtype`.
    Constant { value: Value, dtype: DataType },                                // unchanged
    /// The evaluation time as a value.
    EvaluationTime,                                                            // unchanged
    /// Replaces the listed timestamp columns with the scope's timestamp.
    ScopeTimestamp { columns: Vec<Option<usize>> },                            // unchanged
    /// PromQL instant selection: each series' latest sample within the lookback.
    CurrentSeries { identity: usize, coordinate: usize, value: usize, lookback_ms: i64 },   // unchanged
    /// Concatenates its inputs.
    Union,                                                                     // unchanged
    /// PromQL scalar(): one column of a one-row vector as a scalar.
    VectorToScalar { column: usize },                                          // unchanged
    /// PromQL vector/scalar binary operator.
    VectorBinary { operator: BinaryOperator, return_bool: bool },              // unchanged
    /// Binary operator over rows matched by key columns.
    AlignedBinary { keys: Vec<(usize, usize)>, values: (usize, usize), operator: BinaryOperator },   // unchanged
    /// A PromQL range function over a range vector.
    RangeWindow { intent: Box<AggIntent<ColumnRef>> },                         // unchanged
    /// PromQL histogram_quantile over bucket rows.
    HistogramQuantile,                                                         // unchanged
    /// Per-series window (range function or subquery) at an offset, @ and step.
    SeriesWindow { function: Option<Box<AggIntent<ColumnRef>>>, coordinate: usize, value: usize, range_ms: i64,
        offset_ms: i64, at_ms: Option<i64>, steps: Option<SubquerySteps>, range_at: Option<AtModifier>,
        steps_range_at: Option<AtModifier> },                                  // unchanged
    /// PromQL vector matching labels (on/ignoring, group_left/right).
    SeriesLabels { kind: VectorMatchKind, labels: Vec<String>, unique: bool }, // unchanged
    /// Per-series binary operator; `scalars` marks scalar sides.
    SeriesBinary { operator: BinaryOperator, scalars: [bool; 2] },             // unchanged
    /// PromQL label_replace.
    SeriesRelabel { destination: String, replacement: String, source_regex: Option<(String, String)> },   // unchanged
    /// Per-series histogram_quantile; `quantile` is the f64's bits.
    SeriesHistogramQuantile { quantile: u64, le: usize },                     // unchanged
    /// Computes one output column per expression.
    Project(Vec<Expression>),                                                  // unchanged
    /// Keeps rows where the expression is true.
    Filter(Expression),                                                        // unchanged
    /// Keeps `n` rows after `offset` per group.
    Limit { n: u64, offset: u64, groups: Vec<usize> },                         // unchanged
    /// Sorts rows by `keys` within each group.
    Sort { keys: Vec<SortKey>, groups: Vec<usize> },                           // unchanged
    /// A SQL window function per group over an optional time window.
    Window { intent: Box<AggIntent<ColumnRef>>, coordinate: usize, value: usize, groups: Vec<usize>,
        window: Option<(i64, i64)> },                                          // unchanged
    /// Keeps left rows with a key match on the right.
    SemiJoin { keys: Vec<(usize, usize)>, require_complete_right: bool },      // unchanged
    /// A relational join of `kind` with a predicate.
    Join { kind: JoinKind, predicate: Box<CompiledExpression> },               // unchanged
    /// Reads the top `k` items of a keyed summary state.
    KeyedEvaluation { state: usize, k: usize },                                // unchanged
    /// Merges the summary states in column `state` per group.
    SummaryMerge { state: usize, groups: Vec<usize> },                         // unchanged
    /// Reads one statistic from the summary state in column `state`.
    Evaluation { state: usize, query: SummaryEvaluation },                     // unchanged
}

crates/asap-physical-operators/src/capability.rs

/// Rejects a filtered `SummaryAgg` timed `IngestionTime` (an unsatisfiable plan): ingestion
/// builds read raw samples without the query's columns. Used by precompute and binding.
pub fn validate_ingestion_time_summary<C>(filter: Option<&planner_types::ir::Predicate<C>>) -> Result<(), String>;  // new: a filtered SummaryAgg runs only at query time
/// `(shared_rows, shared_columns, width, depth)` of a HydraCms family; errors (bad plan)
/// unless it is Count-Min with the Hydra's width and depth and valid dimensions.
pub(crate) fn hydra_cms_shape(family: &SummaryFamilyType) -> Result<(usize, usize, usize, usize), String>;          // new: plan contract -> grid dimensions
/// Checks a HydraCms update binds: one item column, and unit weight or a weight column
/// proven non-negative; otherwise the plan is rejected as bad input.
fn validate_hydra_cms(family: &SummaryFamilyType, input: &SummaryUpdate, grouping: &GroupingStrategy) -> Result<(), String>;  // new

crates/types/src/ir/asap.rs, non_asap.rs

// Schema derivation: a quantile over a relation is nullable for a filtered measure,
// a nullable input column or a global reduction, including through SummaryMerge.
/// Whether a quantile read from `state` can have no value, and so has a nullable type.
fn quantile_may_be_empty(state: &OperatorNode) -> bool;   // new (private)

Usage, end to end (the plan a HydraCms SummaryAgg binds, as in tests/hydra_cms_execution.rs; numbers from a scratch run of that harness at 1b2899a5). Pass 1 does not emit this node yet (G7b), so step 1 writes the plan node a planner candidate will carry:

// 1. The question: per job, how many `checkout` events are there? With the 9 input rows
//    (job, service, weight): api has checkout x2, search, auth; batch has checkout, export;
//    db has checkout, vacuum x2.
//    `grouping` is the shared Hydra grid (3 x 64 cells, each a Count-Min of width 64, depth 3)
//    that every job's state reads; `family` is the matching Count-Min family. Both go into the
//    SummaryAgg below and are checked against each other at binding (hydra_cms_shape).
let grouping = GroupingStrategy::SharedMultiSubpopulation {
    kind: HydraKind::HydraCms,
    params: HydraParams::HydraCms { width: 64, depth: 3, shared_rows: 3, shared_columns: 64 },
};
let family = FieldDataType::Sketch(SketchKind::new(SketchAlgorithm::Cms, SketchParams::Cms { width: 64, depth: 3 }), grouping.clone());
// `agg` is the plan's SummaryAgg: item = `service`, weight = Constant(1.0) with a UnitCount
// proof (a row count), reduction = by `job`. `read` is the SummaryEstimate on top of it, a
// valued PointCount { key: service, value: "checkout" }.
let agg = ASAPOp::SummaryAgg { child: 0, family, input: unit_count(), reduction: Reduce(by job), grouping, filter: None };
let read = ASAPOp::SummaryEstimate { summary_input: 1, query: PointCount { key: "service", value: Some("checkout") } };

// 2. Binding. `input` is the input schema (job Utf8, service Utf8, weight Float64).
//    `build` is the compiled SummaryAgg (plan node 1): compile_node validates the contract
//    (validate_hydra_cms: one non-null item column, unit or non-negative weight) and returns
//    Operator::shared_summary_build, a SharedSummaryBuild whose state is
//    create_hydra_cms(&family), one HydraCms for all groups. Step 3 runs it under `read`.
let input = input_schema();
let build = compile_node(&node(1, Payload::ASAP(agg), /* (job, state) */), &[input])?;

// 3. Execution. Each row calls grid.count(job, service, 1). The build emits one row per job,
//    a HydraCmsGroup { grid: Arc<HydraCms> (shared), group: job }; Batch::bytes reserves the
//    grid once through shared_state. `points` holds the estimate node's output, read through
//    HydraCmsGroup::estimate -> grid.point(job, "checkout"): what the user gets.
//    (`run` is the test harness: it compiles `build` and `read` into one DAG and executes it.)
let points = run(unit_count(), None, vec![batch(&ROWS)]);
// -> {"api": 2.0, "batch": 1.0, "db": 1.0}: the exact counts here; the bound allows up to
//    e·9·(1/64 + 1/64) ≈ 0.76 over them.

// 4. The same build over a split stream (rows 0..5 and 5..9), merged with SummaryMerge
//    (HydraCms::merge, once per pair of grids). `merged` holds that answer; it equals `points`.
let merged = run(unit_count(), None, vec![batch(&ROWS[..5]), batch(&ROWS[5..])]);
// -> {"api": 2.0, "batch": 1.0, "db": 1.0}

// 5. With the `weight` column as the weight (non-negative proof), each point is a weight sum.
//    `bytes` holds it: api's checkout weights 3 + 2, batch's 7, db's 1.
let bytes = run(weighted(non_negative()), None, vec![batch(&ROWS)]);
// -> {"api": 5.0, "batch": 7.0, "db": 1.0}

Before this PR

Using the table events(g, x, y): group a has x values 1, -1, 2; b has -3, 0; c has 5.

plan	result
SELECT g, COUNT(*) FILTER (WHERE x > 0) … GROUP BY g (exact, through the planner pipeline)	filtered summary update has no native implementation
SELECT g, SUM(x) FILTER (WHERE x > 0) … GROUP BY g	same error
SELECT g, COUNT(y) … GROUP BY g (nullable y)	same error
SummaryAgg with HydraCms grouping (count or weighted)	shared summary grouping has no registered kernel

After this PR

plan	result
filtered COUNT(*)	a 2, b 0, c 1
filtered SUM(x)	a 3.0, b NULL, c 5.0
COUNT(y)	a 1, b 0, c 1
HydraCms count of service="checkout" by (job)	one shared grid. Each estimate is at least the exact count and within the CMS bound. A split-stream build merged with SummaryMerge equals one build over the whole stream.
HydraCms with a non-negative weight column	each group's item weight sum, at least the exact sum and within the CMS bound over the total weight
HydraCms with an unknown-sign weight, or a negative or fractional row weight	rejected with a clear error
HydraCms group total (a PointCount with no item)	rejected: HydraCms does not answer a group total: asap_sketchlib's Hydra has no group-mass query …
SELECT g, approx_percentile_cont(y, 0.5) … GROUP BY g, nullable y, KLL candidate	the all-NULL group reads NULL instead of failing the query
ungrouped KLL quantile over no row	one NULL row
HydraCms through SummaryMerge with 400 groups	fits the default memory limit, with one merged grid shared by all groups
filtered SummaryAgg timed IngestionTime	rejected: a filtered summary update runs only at query time

Limitations and follow-ups

• No Hydra group total. A group's total (count by (g) read from the Hydra grid) needs a group-mass query, and asap_sketchlib's Hydra has none. executor: HydraCms kernel for count and point queries #600 mirrored sketchlib's private subkey encoding to get it. That would be a local copy, so this PR rejects the query with an explicit error. Group totals will come from an upstream sketchlib API and a move of the pin, as a separate task.
• Integer weights only. Hydra adds i32 weights. Fractional weights, such as a PromQL rate, cannot be added and fail the build, and the total weight of one grid is capped at i32::MAX.
• Pass 1 offering HydraCms, and its PruneReasons, are G7b. In G7b, SQL COUNT(*) GROUP BY k with one Count-Min per group is pruned in Pass 1, so this PR does not change the keyed build for Constant weights.
• Mergeability comes from the family-capability table (feat(types): family-capability table and executor tumbling pane merges #696). That table lists HydraCms as mergeable (same-shape grids add cell-wise) and HydraKll as not. SummaryMerge binding checks it, through require_sound_merge, before the kernel merges. The kernel adds only runtime checks: the same shape, the same group, and the i32 total.
• Out of scope: Hydra top-k (a separate PR after G7) and HydraKll.
• Planner-side timing validation does not yet apply the query-time-only rule. This covers materialization timings and Stage 2 eligibility. Until G8 adds a planner check or a PruneReason, a plan placing a filtered SummaryAgg at ingestion is accepted by the planner and rejected only when the executor compiles it.
• Some workspaces may undercount shared grids. Row-by-row workspaces (row_bytes, e.g. in joins) do not count a shared grid. Hydra states flow only into SummaryMerge and SummaryEstimate, which do count it.
• Merged grids stay reserved until the merge ends. A grid that is replaced inside one SummaryMerge remains reserved until that merge finishes.

Verification

• Tests first. These fail on main with the errors in the Before table:
  • integration-tests/tests/filtered_aggregates.rs: exact filtered COUNT, filtered SUM and COUNT(nullable) through the planner pipeline and the executor, plus hand-filtered exact Count and KLL SummaryAggs.
  • physical_semantics::filtered_summary_build_is_rejected_at_ingestion_time.
• New executor tests:
  • tests/hydra_cms_execution.rs: point counts within the CMS bound, merge equals one build, a filtered Hydra build, and rejection of group totals;
  • many_groups_merge_one_grid_pair: 400 groups in both builds, merged within the default memory limit. It failed first.
  • filtered_aggregates::kll_quantile_of_all_null_group_is_null and global_kll_quantile_over_no_row_is_null. Both failed first.
  • physical_semantics::measure_filters_are_rejected_over_promql_series_identity. It failed first.
  • weighted Hydra tests, written to fail first: weighted_hydra_cms_sums_item_weights_per_group (bounded sums, plus a split-stream merge), unknown_sign_weight_is_rejected and invalid_weight_values_fail_the_build (negative, fractional, NaN, infinite);
  • kernel unit tests in hydra_cms.rs and factory.rs;
  • physical_semantics tests for measure filters, filtered summary builds and serde round trips;
  • the precompute rejection in precompute_population.
• On head 79792ef894f7682896ec8c2c9afa3801ec36e83e, rebased on main 3a25450d (feat(types): family-capability table and executor tumbling pane merges #696):
  • cargo fmt --all -- --check: pass;
  • cargo clippy --workspace --all-targets --all-features --locked -- -D warnings: pass;
  • cargo nextest run --workspace: 1690 passed, 46 skipped.
• Screenshots: not applicable (no visual output).