refactor!: name summary builds by shape; one RowWeight for keyed builds

[zzylol]

Why

Two review comments on #753:

SharedSummaryBuild just reading the name, I cannot tell what it is for

This seems to be a special case, why hydra row weight is a separate enum definition?

The executor had three summary builds named SummaryBuild, KeyedSummaryBuild and SharedSummaryBuild. The names don't say what each one holds, and they don't read well next to each other. The row weight was also a Hydra-only type (HydraRowWeight), while the keyed per-group build (CMS or Count Sketch with heap) took a bare value: usize that had to be a non-null Float64 column. A keyed heap over a row count (Constant(1.0), the unit count) therefore either failed at query time ("keyed summary weight must be a finalized value column") or, in precompute, projected a column of ones.

What

• The three builds are renamed by shape, and so are their constructors:
  • PerGroupSummaryBuild / per_group_summary_build: one state per group, updated from one value column.
  • PerGroupKeyedSummaryBuild / per_group_keyed_summary_build: one keyed state per group whose items take a row weight, e.g. a CMS with heap.
  • AllGroupsSummaryBuild / all_groups_summary_build: one state shared by all groups and emitted per group. Today that is HydraCms.
• HydraRowWeight becomes RowWeight { Column(usize), Constant(NonZeroU32) }. Both keyed builds now use it. With it, the keyed per-group build also accepts:
  • a constant weight, e.g. 1 for a row count;
  • an Int64 weight column. The kernel's weights are f64, so an Int64 value beyond 2^53 is refused at run time rather than rounded;
  • a nullable weight column. A NULL adds nothing and the group is kept, the same as in the all-groups build.
• The query-time physical planner now binds a keyed heap over the unit count with RowWeight::Constant(1). It used to refuse this.
• The raw-sample precompute now adds a keyed build's constant weight directly, without projecting a value column of ones.
• The i32::MAX bound is Hydra-specific. It is checked by the HydraCms kernel check (validate_hydra_cms) and by all_groups_summary_build, not by the keyed per-group build. hydra_constant_weight is renamed to constant_row_weight, because both builds use it now. It bounds a constant by u32::MAX.
• Serialized format (breaking, no backward compatibility):
  • The executor Operator JSON kind tags change: SummaryBuild → PerGroupSummaryBuild, KeyedSummaryBuild → PerGroupKeyedSummaryBuild, SharedSummaryBuild → AllGroupsSummaryBuild.
  • The keyed build's "value": <column> field is replaced by "row_weight": {"Column": <column>} or {"Constant": n}.
  • I found no devtools, viewer or fixture files that use these names. The only hits were Rust tests, which this PR updates. The Operator::name() string stays "SummaryAgg" for all three.

How

• operators/summary/mod.rs:
  • RowWeight replaces HydraRowWeight.
  • per_group_keyed_summary_build(input, family, items, row_weight, groups) checks that a weight column is Float64 or Int64.
  • build_per_group_keyed_summary reads each row's weight from the constant, or from the column through the new keyed_weight. NULL skips the row; an Int64 value must be within ±2^53.
• physical_planner/mod.rs:
  • A private row_weight(input, &update.weight) turns a validated SummaryUpdate.weight into a RowWeight. A column becomes Column(index); a constant goes through constant_row_weight and becomes Constant.
  • Both keyed builds call it.
  • The private summary_build dispatcher is renamed summary_agg_build, so it does not read as one of the three builds.
• physical_planner/precompute.rs: for a keyed update with a constant weight, the projection drops the value column and the build gets RowWeight::Constant. A non-integer keyed constant is now refused there. Stage 1 only produces Constant(1.0) for keyed updates, so no planner output is affected.
• capability.rs:
  • validate_hydra_cms keeps the "HydraCms" prefix on constant-weight errors.
  • After constant_row_weight, it also refuses a constant above i32::MAX with the same error as the constructor.
  • The no-kernel error now reads "a shared-grouping (Hydra) family has no registered kernel".

Key code interfaces added / modified

crates/executor/src/operators/summary/mod.rs

/// What each input row adds to its item's count in a keyed summary build
/// (per_group_keyed_summary_build or all_groups_summary_build). Produced by the
/// physical planner from a validated `SummaryUpdate.weight`. The type stays, renamed.
pub enum RowWeight {                                   // was HydraRowWeight
    /// The value of this input column (Float64 or Int64); a NULL adds nothing.
    Column(usize),                                     // unchanged
    /// This positive integer per row, e.g. 1 for a row count; the all-groups
    /// (HydraCms) build refuses one above i32::MAX.
    Constant(NonZeroU32),                              // unchanged
}

impl Operator {
    /// One summary state per group, updated from one value column per row
    /// (`time` orders counter samples): an exact aggregate, a quantile or
    /// distinct sketch, or UnivMon over that column's values.
    pub fn per_group_summary_build(                    // was summary_build
        input: SchemaRef,
        family: SummaryFamilyType,
        value: usize,
        time: Option<usize>,
        groups: Vec<usize>,
    ) -> Result<Self, Error>;

    /// One keyed state per group (CMS or Count Sketch with heap): each row adds
    /// its `row_weight` to its `items` tuple. Err(Invalid) for a weight column
    /// that is not Float64 or Int64 (bad input).
    pub fn per_group_keyed_summary_build(              // was keyed_summary_build
        input: SchemaRef,
        family: SummaryFamilyType,
        items: Vec<usize>,                             // unchanged, moved before the weight
        row_weight: RowWeight,                         // was value: usize (a non-null Float64 column)
        groups: Vec<usize>,
    ) -> Result<Self, Error>;

    /// One state shared by all groups (a HydraCms grid), emitted as one row per
    /// group. Err(Invalid) for a constant weight above i32::MAX (bad input).
    pub fn all_groups_summary_build(                   // was shared_summary_build
        input: SchemaRef,
        family: SummaryFamilyType,
        item: usize,
        row_weight: RowWeight,                         // was HydraRowWeight
        groups: Vec<usize>,
    ) -> Result<Self, Error>;
}

crates/executor/src/operators/mod.rs

/// The serialized `kind` of an executor `Operator` (crate-private type, public
/// wire format). The type stays; three variants are renamed.
enum Kind {
    /// In-memory input batches; never serialized.
    Source(Vec<Batch>),                                          // unchanged
    /// A constant scalar.
    Constant {                                                   // unchanged
        /// The constant value.
        value: Value,                                            // unchanged
        /// Its type.
        dtype: DataType,                                         // unchanged
    },
    /// The query's evaluation time as a scalar.
    EvaluationTime,                                              // unchanged
    /// Stamp each row with the scope's timestamp.
    ScopeTimestamp {                                             // unchanged
        /// For each output column, the input column it copies (None = the timestamp).
        columns: Vec<Option<usize>>,                             // unchanged
    },
    /// The latest sample of each series within a lookback.
    CurrentSeries {                                              // unchanged
        /// The series identity column.
        identity: usize,                                         // unchanged
        /// The timestamp column.
        coordinate: usize,                                       // unchanged
        /// The sample value column.
        value: usize,                                            // unchanged
        /// The lookback, in milliseconds.
        lookback_ms: i64,                                        // unchanged
    },
    /// Concatenate inputs of one schema.
    Union,                                                       // unchanged
    /// PromQL `scalar()`.
    VectorToScalar {                                             // unchanged
        /// The value column read.
        column: usize,                                           // unchanged
    },
    /// PromQL vector/scalar binary operator.
    VectorBinary {                                               // unchanged
        /// The operator.
        operator: BinaryOperator,                                // unchanged
        /// PromQL `bool` modifier.
        return_bool: bool,                                       // unchanged
    },
    /// Binary operator over key-aligned inputs.
    AlignedBinary {                                              // unchanged
        /// (left, right) key column pairs.
        keys: Vec<(usize, usize)>,                               // unchanged
        /// (left, right) value columns.
        values: (usize, usize),                                  // unchanged
        /// The operator.
        operator: BinaryOperator,                                // unchanged
    },
    /// A PromQL range-vector function.
    RangeWindow {                                                // unchanged
        /// The function applied over each range.
        intent: Box<AggIntent<ColumnRef>>,                       // unchanged
    },
    /// PromQL `histogram_quantile`.
    HistogramQuantile,                                           // unchanged
    /// A per-series window function or subquery.
    SeriesWindow {                                               // unchanged
        /// The function applied; None = select samples only.
        function: Option<Box<AggIntent<ColumnRef>>>,             // unchanged
        /// The timestamp column.
        coordinate: usize,                                       // unchanged
        /// The sample value column.
        value: usize,                                            // unchanged
        /// The window range, in milliseconds.
        range_ms: i64,                                           // unchanged
        /// PromQL `offset`, in milliseconds.
        offset_ms: i64,                                          // unchanged
        /// PromQL `@`, in milliseconds; None = none.
        at_ms: Option<i64>,                                      // unchanged
        /// Subquery steps; None = not a subquery.
        steps: Option<SubquerySteps>,                            // unchanged
        /// `@ start()`/`@ end()` of the range; default None.
        range_at: Option<AtModifier>,                            // unchanged
        /// `@ start()`/`@ end()` of the subquery steps; default None.
        steps_range_at: Option<AtModifier>,                      // unchanged
    },
    /// Per-series label matching (`on`/`ignoring`, `group_left/right`).
    SeriesLabels {                                               // unchanged
        /// The vector match kind.
        kind: VectorMatchKind,                                   // unchanged
        /// The matched labels.
        labels: Vec<String>,                                     // unchanged
        /// Whether each match must be unique.
        unique: bool,                                            // unchanged
    },
    /// Per-series binary operator.
    SeriesBinary {                                               // unchanged
        /// The operator.
        operator: BinaryOperator,                                // unchanged
        /// Which of the two sides is a scalar.
        scalars: [bool; 2],                                      // unchanged
    },
    /// PromQL `label_replace`.
    SeriesRelabel {                                              // unchanged
        /// The destination label.
        destination: String,                                     // unchanged
        /// The replacement template.
        replacement: String,                                     // unchanged
        /// (source label, regex); None = constant replacement.
        source_regex: Option<(String, String)>,                  // unchanged
    },
    /// Per-series histogram quantile.
    SeriesHistogramQuantile {                                    // unchanged
        /// `f64` bits: JSON cannot encode the NaN and infinite quantiles.
        quantile: u64,                                           // unchanged
        /// The `le` bucket-bound column.
        le: usize,                                               // unchanged
    },
    /// A projection.
    Project(Vec<Expression>),                                    // unchanged
    /// A row filter.
    Filter(Expression),                                          // unchanged
    /// Per-group limit and offset.
    Limit {                                                      // unchanged
        /// Rows kept per group.
        n: u64,                                                  // unchanged
        /// Rows skipped per group.
        offset: u64,                                             // unchanged
        /// The group columns.
        groups: Vec<usize>,                                      // unchanged
    },
    /// Per-group sort.
    Sort {                                                       // unchanged
        /// The sort keys.
        keys: Vec<SortKey>,                                      // unchanged
        /// The group columns.
        groups: Vec<usize>,                                      // unchanged
    },
    /// A windowed aggregate.
    Window {                                                     // unchanged
        /// The aggregate applied.
        intent: Box<AggIntent<ColumnRef>>,                       // unchanged
        /// The timestamp column.
        coordinate: usize,                                       // unchanged
        /// The value column.
        value: usize,                                            // unchanged
        /// The group columns.
        groups: Vec<usize>,                                      // unchanged
        /// The fixed (start, end) window in ms; None = the scope's.
        window: Option<(i64, i64)>,                              // unchanged
    },
    /// A grouped aggregate.
    Aggregate {                                                  // unchanged
        /// The group columns.
        groups: Vec<usize>,                                      // unchanged
        /// The measures computed.
        measures: Vec<Reduction>,                                // unchanged
        /// Per-measure row filters (SQL `FILTER (WHERE …)`); default empty.
        filters: Vec<Option<Expression>>,                        // unchanged
    },
    /// A semi-join.
    SemiJoin {                                                   // unchanged
        /// (left, right) key column pairs.
        keys: Vec<(usize, usize)>,                               // unchanged
        /// Whether the right input must be complete.
        require_complete_right: bool,                            // unchanged
    },
    /// A relational join.
    Join {                                                       // unchanged
        /// Inner, left, ...
        kind: JoinKind,                                          // unchanged
        /// The join predicate.
        predicate: Box<CompiledExpression>,                      // unchanged
    },
    /// One summary state per group, updated from one value column per row.
    PerGroupSummaryBuild {                                       // was SummaryBuild
        /// The summary family built.
        family: SummaryFamilyType,                               // unchanged
        /// The value column each row updates with.
        value: usize,                                            // unchanged
        /// The timestamp column that orders counter samples; None = none.
        time: Option<usize>,                                     // unchanged
        /// The group-by columns.
        groups: Vec<usize>,                                      // unchanged
        /// Rows for which this is not true update no state; default None.
        filter: Option<Box<Expression>>,                         // unchanged
    },
    /// One keyed state per group whose items take a row weight (CMS/CS with heap).
    PerGroupKeyedSummaryBuild {                                  // was KeyedSummaryBuild
        /// The keyed sketch family built.
        family: SummaryFamilyType,                               // unchanged
        /// The item identity columns.
        items: Vec<usize>,                                       // unchanged
        /// What each row adds to its item's count.
        row_weight: RowWeight,                                   // was value: usize
        /// The group-by columns.
        groups: Vec<usize>,                                      // unchanged
        /// Rows for which this is not true update no state; default None.
        filter: Option<Box<Expression>>,                         // unchanged
    },
    /// One state shared by all groups and emitted per group (HydraCms today).
    AllGroupsSummaryBuild {                                      // was SharedSummaryBuild
        /// The HydraCms family built.
        family: SummaryFamilyType,                               // unchanged
        /// The item column.
        item: usize,                                             // unchanged
        /// What each row adds to its item's count.
        row_weight: RowWeight,                                   // was HydraRowWeight
        /// The group-by columns.
        groups: Vec<usize>,                                      // unchanged
        /// Rows for which this is not true update no state; default None.
        filter: Option<Box<Expression>>,                         // unchanged
    },
    /// Read a keyed state's top-k items.
    KeyedEvaluation {                                            // unchanged
        /// The state column.
        state: usize,                                            // unchanged
        /// Items read per group.
        k: usize,                                                // unchanged
    },
    /// Merge summary states per group.
    SummaryMerge {                                               // unchanged
        /// The state column.
        state: usize,                                            // unchanged
        /// The group columns.
        groups: Vec<usize>,                                      // unchanged
    },
    /// Evaluate a summary state.
    Evaluation {                                                 // unchanged
        /// The state column.
        state: usize,                                            // unchanged
        /// The sketch query or exact evaluation.
        query: SummaryEvaluation,                                // unchanged
    },
}

crates/executor/src/capability.rs

/// A `SummaryUpdate` constant weight as the positive integer a
/// RowWeight::Constant holds (at most u32::MAX); Err (bad input) otherwise.
/// Used by the physical planner and precompute for both keyed builds.
pub(crate) fn constant_row_weight(weight: f64) -> Result<NonZeroU32, String>; // was hydra_constant_weight (bounded by i32::MAX)

Usage (end to end)

This is the new test count_ranked_heap_precompute_adds_a_constant_row_weight in crates/integration-tests/tests/precompute_raw_samples.rs. It passes at the linked commit.

// The query the user writes: the 2 services with the most samples, at ε = 0.02.
let query = "topk(2, count by (service) (m))";
// The fixture's raw samples: 5 series × 5 samples, services "a" (2 series),
// "b" (2 series) and no service (1 series); checked against below.
let rows = samples();
// Stage 1 logical candidates for the query (frontend lowering + candidate
// enumeration), one of which builds a Count-Min heap.
for candidate in candidates(query, AccuracyTarget::max_error(0.02)) {
    // The candidate exported as a PhysicalASAPDAG with its summaries placed at
    // ingestion (MatAtIngestion), so the heap reads raw samples.
    let dag = compile_at_ingestion_physical_asap_dag(&candidate).unwrap();
    // Each raw-input SummaryAgg: (raw source id, summary id).
    for (source, root) in raw_summaries(&dag) {
        // ... keep the CmsWithHeap SummaryAgg; its update is the unit count:
        assert_eq!(input.weight, SummaryInputExpr::Constant(1.0));
        // The executor precompute DAG for that summary, as JSON. The keyed build
        // is PerGroupKeyedSummaryBuild with row_weight {"Constant": 1}, and the
        // projection below it has no column of ones.
        let compiled = serde_json::to_value(precompute::compile(&dag, &[source], &[root]).unwrap())
            .unwrap()
            .to_string();
        assert!(compiled.contains(r#""row_weight":{"Constant":1}"#));
        // Run it over the samples: the heap's per-service counts equal the exact
        // sample counts, a: 10, b: 10, (no service): 5.
        assert_eq!(check(query, &dag, source, root, &rows), Ok("CmsWithHeap".into()));
    }
}

Before this PR

• The names SummaryBuild, KeyedSummaryBuild and SharedSummaryBuild didn't say which build holds one state per group and which holds one state for all groups.
• HydraRowWeight existed for HydraCms only.
• The keyed per-group build required a non-null Float64 weight column. At query time, a count-ranked heap (Constant(1.0)) failed to bind with "keyed summary weight must be a finalized value column". In precompute, it projected FiniteFloat64(1.0) as a value column on every row.

After this PR

• The three builds are named by shape and documented on their own.
• One RowWeight serves both keyed builds.
• The keyed per-group build takes a constant weight, an Int64 column or a nullable column. The Hydra-only i32::MAX bound stays in the all-groups constructor.
• The query-time count-ranked heap binds.
• The precompute no longer projects the constant.

Verification

Each new test failed before the change:

• query_time_count_heap_binds_a_constant_row_weight failed with "keyed summary weight must be a finalized value column".
• raw_sample_heaps_resolve_items_from_labels's new assertion failed with the old precompute behaviour (constant projected as a column).

New tests in crates/executor/tests/keyed_summary_row_weight.rs:

• query_time_count_heap_binds_a_constant_row_weight: compile a SummaryAgg CmsWithHeap with Constant(1.0) → row_weight {"Constant":1}, counts api/checkout 2, api/search 1, batch/export 3.
• constant_row_weight_equals_a_column_of_that_constant: Constant(3) equals Column over Float64 3.0, both 3× the counts.
• int64_weight_column_equals_float64_column: Int64 2 gives 2× the counts. 2^53 + 1 is refused with "no exact Float64", and a Utf8 weight is refused at construction.
• int64_weight_column_equals_float64_column also checks that exactly 2^53 builds on the Count-Min heap and −2^53 builds on the Count Sketch heap (signed weights).
• null_weight_adds_nothing: NULL weights add no items, and both groups keep a state.

New test in crates/executor/tests/hydra_cms_execution.rs:

• constant_weight_above_i32_is_refused_at_validation: Constant(2147483648.0) + PositiveConstant is refused with the "HydraCms … i32::MAX" error by validate_summary_kernel and by compile_node. It failed before the validation fix.

Integration test:

• count_ranked_heap_precompute_adds_a_constant_row_weight (the Usage above), plus a new assertion in raw_sample_heaps_resolve_items_from_labels.

Commands run:

• cargo nextest run -p asap-executor --lib --test hydra_cms_execution --test physical_dag --test physical_semantics --test univmon_execution --test weighted_topk_binding --test current_series_heap --test blocking_resources --test keyed_summary_row_weight --test plan_properties: 173 passed, 4 skipped (pre-existing ignores).
• cargo nextest run -p asap-integration-tests --test precompute_raw_samples --test kll_pane_execution: 6 passed.
• All of the above were rerun after rebasing onto main 4ef9f3e5.
• cargo fmt --all -- --check and cargo clippy -p asap-executor -p asap-integration-tests --all-targets --all-features --locked -- -D warnings: clean.

Limitations and follow-ups

• Markdown docs still use the old names; Markdown doc updates are tracked in Update the docs after the G1–G11 consolidation groups land #677:
  • docs/develop_docs/physical-compile-coverage.md row 26 says keyed_summary_build;
  • docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md:200 says SummaryBuild.
• feat(planner): plan_installable, the one call a deployment runtime makes (#749) #755 also edits crates/executor/src/physical_planner/mod.rs. Whichever PR merges second rebases.