refactor: Hydra row weight enum, TooManyCandidates count, cost_unit rename, stacked-shift pane fix

[zzylol]

Why

The review of #752 found interfaces that let a caller say something meaningless, plus names that do not say what they hold:

• Operator::shared_summary_build(.., weight: Option<usize>, constant_weight: i32, ..) took a weight column and a constant weight as two parameters. With weight = Some(col) and constant_weight = 2, the build silently ignored the 2. A runtime constant_weight <= 0 check was needed because the type allowed both zero and negative values.
• PerComputationCandidateError::TooManyCandidates { candidates, max }: the candidates field held a description such as "rollup candidates", not candidates. The message also never said how many were found.
• CandidateCost.unit held the unit of cost_per_second, but its name did not say so.
• pane_window_ms checked for an @ anchor only on the shift directly below the range. If an anchored shift sat under an offset shift, the build was offered panes.

What

1. Hydra row weight. A shared HydraCms build now takes one row_weight: HydraRowWeight, which is either Column(usize) or Constant(NonZeroU32). A column and a constant can no longer be passed together. shared_summary_build refuses a constant above i32::MAX (asap_sketchlib's Hydra adds i32 weights). This applies to direct calls and to deserialized documents. The <= 0 check is removed because the type rules it out.
2. TooManyCandidates { counted, found, max }, with the message found {found} {counted}, more than {max}. Each of the five cap sites passes the count it found.
3. CandidateCost.unit → cost_unit. The stage_pipeline JSON key, the viewer readers and the committed example change with it.
4. Nits.
   • The validate_hydra_cms doc now names the UnitCount and PositiveConstant proofs.
   • The physical planner's leftover _ => (None, 1) weight arm is now unreachable!("validated HydraCms weight"). SummaryInputExpr also has Tuple and EntityIdentity, so the match cannot be made total.
   • pane_window_ms now walks every stacked TimeShift below the range and refuses panes when any of them has an @ anchor.

Serialized format changes

• Executor Operator JSON, SharedSummaryBuild: the fields weight (optional column index) and constant_weight (default 1) are replaced by one required field, row_weight. Its value is {"Constant": 2} or {"Column": 2}. Old documents are not readable; keeping them readable would need a custom deserializer, and the repo has no backward-compatibility rule.
• stage_pipeline document, stage3_plan_selection.costs.<id>: the key unit is now cost_unit. The viewer validates cost_unit, and old viewer documents fail validation with cost_unit must be a string. The committed tools/dag-viewer/examples/planner-layering-example1.json is regenerated; only the 19 renamed keys change.

How

• HydraRowWeight lives next to the shared summary build in operators/summary and is re-exported from operators. The physical planner maps the validated SummaryUpdate.weight to it: a Column resolves to its input index, and a Constant goes through capability::hydra_constant_weight. That function now returns a NonZeroU32 that is at most i32::MAX, instead of an i32. The build reads row_weight once per row: a constant is added as is, a column value is added, and a NULL value adds nothing.
• Kind::SharedSummaryBuild stores row_weight directly. Deserialization goes through UncheckedOperator → shared_summary_build, so the i32::MAX check also covers documents.
• The capability sharing cap reports the size of the largest group. The pane cap reports the number of pane candidates, subsets reports n, and the combined reuse cap reports the rollup count plus the top-k count.
• The viewer's Stage 3 candidate rows drop their unit field. No code read it; the lane header and ranking label read cost.cost_unit from the cost itself.

Key code interfaces added / modified

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

summary/mod.rs · mod.rs

/// What each row of a shared HydraCms build adds to its item's count.
/// Produced by the physical planner from a validated `SummaryUpdate` weight;
/// read once per row by the build. Re-exported as `operators::HydraRowWeight`.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum HydraRowWeight {                                     // new
    /// The value of this input column (Float64 or Int64, non-negative);
    /// a row whose value is NULL adds nothing.
    Column(usize),                                            // new (was `weight: Some(column)`)
    /// This positive integer per row (1 for a row count), at most `i32::MAX`;
    /// `shared_summary_build` refuses a larger one.
    Constant(NonZeroU32),                                     // new (was `constant_weight: i32`)
}

impl Operator {
    /// One HydraCms state shared by every group, emitted as one row per group
    /// that reads it; each row adds `row_weight` to its `item` value. Errors
    /// (bad input) when the constant is above `i32::MAX`, the weight column is
    /// not Float64/Int64, or the item is not a non-null Utf8/Int64/Bool column.
    pub fn shared_summary_build(
        input: SchemaRef,
        family: SummaryFamilyType,
        item: usize,
        row_weight: HydraRowWeight,                           // was `weight: Option<usize>, constant_weight: i32`
        groups: Vec<usize>,
    ) -> Result<Self, Error>;
}

/// One variant of the private `enum Kind`, whose JSON is the executor's
/// operator wire format. The enum and its other variants stay unchanged;
/// only this variant's fields change.
Kind::SharedSummaryBuild {
    /// The per-group Count-Min family the shared Hydra emulates.
    family: SummaryFamilyType,                                // unchanged
    /// The input column whose values are counted.
    item: usize,                                              // unchanged
    /// What each row adds; JSON `{"Constant": 2}` or `{"Column": 2}`. Required.
    row_weight: HydraRowWeight,                               // new
    /// The weight column index; absent for a constant weight.
    weight: Option<usize>,                                    // removed
    /// The per-row constant without a weight column; default 1.
    constant_weight: i32,                                     // removed
    /// The input columns that key each group.
    groups: Vec<usize>,                                       // unchanged
    /// Rows for which this is not true update no state; their group is kept.
    filter: Option<Box<Expression>>,                          // unchanged
}

/// Removed: `fn unit_weight() -> i32` and `fn is_unit_weight(&i32) -> bool`,
/// the serde default and skip helpers for `constant_weight`.

crates/executor/src/capability.rs

capability.rs

/// A HydraCms constant weight as a positive integer at most `i32::MAX`;
/// an error (bad input) for zero, negative, fractional, non-finite or too
/// large weights. Used by binding validation and the physical planner.
pub(crate) fn hydra_constant_weight(weight: f64) -> Result<NonZeroU32, String>; // was -> Result<i32, String>

crates/logical-asap-optimization/src/per_computation_candidates.rs

per_computation_candidates.rs

/// Why Stage 1 could not list a computation's or a Pass 2 rule's candidates.
/// The type stays; only `TooManyCandidates` changes.
pub enum PerComputationCandidateError {
    /// A Pass 2 rule found more than `max` candidates of one kind, whose 2^n
    /// subsets it would enumerate: the workload is too large to plan this way,
    /// not bad input. Message: "found {found} {counted}, more than {max}".
    TooManyCandidates {
        /// What was counted, e.g. "rollup candidates".
        counted: &'static str,                                // was `candidates`
        /// How many were found, more than `max`.
        found: usize,                                         // new
        /// The cap that was exceeded.
        max: usize,                                           // unchanged
    },
    /// The logical DAG's schema could not be derived (internal bug or bad IR).
    Structure(SchemaDerivationError),                         // unchanged
    /// The Stage 1 input already carries timing or materialization (caller bug).
    AssignedPhysicalAnnotation,                               // unchanged
    /// The Stage 1 input already contains ASAP operators (caller bug).
    ContainsAsap,                                             // unchanged
    /// An approximate target's epsilon or delta is outside (0, 1) (bad input).
    InvalidAccuracy { root: Option<String>, measure: Option<usize> }, // unchanged
    /// A choice does not name one listed candidate per target (caller bug).
    InvalidChoice,                                            // unchanged
    /// A substitution Stage 1 does not support (unsatisfiable request).
    Unsupported(&'static str),                                // unchanged
}

crates/plan-selection/src/lib.rs, crates/devtools/src/bin/stage_pipeline.rs, tools/dag-viewer/{stages,app,editor}.js

lib.rs · stage_pipeline.rs · stages.js

/// One candidate's price under the selected cost model. The type stays.
pub struct CandidateCost {
    /// The candidate's resource usage (CPU, memory, disk, latency per query).
    pub usage: ResourceUsage,                                 // unchanged
    /// Its price per second under cost model 1 or 2; `None` under cost model 3.
    pub cost_per_second: Option<f64>,                         // unchanged
    /// The unit of `cost_per_second`: always `COST_UNIT` ("cost_per_second").
    pub cost_unit: &'static str,                              // was `unit`
    /// Where its prices came from (formulas, measured, or mixed).
    pub source: &'static str,                                 // unchanged
    /// Each physical plan DAG node's price, by node id.
    pub per_node: BTreeMap<PhysicalASAPNodeId, NodeCost>,     // unchanged
}

// stage_pipeline JSON: stage3_plan_selection.costs.<id>
{ "cost_per_second": 0.000132, "cost_unit": "cost_per_second", /* was "unit" */ "source": "...", "usage": {...}, "per_node": {...} }
// Viewer Stage 3 candidate rows (stageCandidateRows): `unit` removed (unread).

Usage (end to end)

No workload reaches a HydraRowWeight::Constant today: the Stage 1 rules emit only UnitCount, ResetAwareCounterDerivative and CounterSamples weight proofs, never PositiveConstant. So the Hydra flow below starts at the Stage 2 SummaryAgg node that a future rule would produce; the cost_unit flow starts from a real stage_pipeline run.

Shared HydraCms build with a constant weight. This is the flow of row_weight_serializes_as_one_field_and_rejects_constants_above_i32 and positive_constant_weight_adds_the_constant_per_row in crates/executor/tests/hydra_cms_execution.rs, both run at the head commit:

// The Stage 2 export's SummaryAgg node: a HydraCms (width 64, depth 3, 3x64
// shared grid) grouped by `job` (column 0) that counts `service` (column 1),
// each row adding the constant weight 2 (proof: PositiveConstant).
let update = SummaryUpdate {
    item: Some(SummaryInputExpr::Column(ColumnRef::Named("service".into()))),
    weight: SummaryInputExpr::Constant(2.0),
    weight_domain: WeightDomain::NonNegative { proof: NonNegativeWeightProof::PositiveConstant },
};
// The executor's binding step: validate_hydra_cms accepts the weight, and the
// physical planner maps Constant(2.0) through hydra_constant_weight to
// HydraRowWeight::Constant(2), then calls Operator::shared_summary_build.
let build: Operator = compile_node(&summary_agg(update, None), &[input_schema()])?;
// The compiled operator as stored in a compiled physical DAG document.
let wire: serde_json::Value = serde_json::to_value(&build)?;
assert_eq!(wire["kind"]["SharedSummaryBuild"]["row_weight"], json!({"Constant": 2}));
// A weight column instead (`weight` = column 2) serializes as {"Column": 2}.
// Running the build over the 9 test rows and reading each job's `checkout`
// point count gives exactly twice the unit-count answer for api, batch, db.
// A document edited to {"Constant": 2147483648} fails to deserialize:
// "HydraCms constant weight must be at most i32::MAX ... got 2147483648".

The cost unit as stage_pipeline users see it (run at the head commit):

$ stage_pipeline --promql "topk by (job) (10, rate(x[1m]))" --max-error 0.01 --failure-probability 0.001 --out run.json
# stage3_plan_selection: selected "P1"; costs.P1 =
#   {"cost_per_second": 0.00013194444444444443, "cost_unit": "cost_per_second",
#    "source": "analytical-cost-v2 (illustrative statistics)", ...}

Before this PR

• shared_summary_build(input, family, item, Some(2), 5, groups) built a column-weighted Hydra and dropped the 5 without an error. constant_weight = 0 was rejected only at runtime.
• The reuse cap error read more than 20 rollup and top-k reuse candidates together (2^n Pass 2 candidates), without the count found.
• The cost JSON had "unit": "cost_per_second".
• SummaryAgg over TimeRange(5m) over TimeShift(offset 1m) over TimeShift(@ 100) was offered panes (pane_window_ms = Some(300000)).

After this PR

• The Hydra row weight is one value that cannot be both a column and a constant. A constant above i32::MAX is refused at construction and at deserialization.
• found 21 rollup and top-k reuse candidates together, more than 20.
• "cost_unit": "cost_per_second".
• An @ anchor anywhere in the stacked shifts below the range means no panes (None).

Verification

• Failing test first for the stacked-shift fix: asap_aware_cse::tests::an_anchor_below_stacked_shifts_has_no_panes failed before the fix (left: Some(300000), right: None) and passes after it.
• cargo nextest run -p asap-executor -E 'test(/hydra|shared|constant_weight/) | binary(hydra_cms_execution)': 21 passed. This includes the new row_weight_serializes_as_one_field_and_rejects_constants_above_i32.
• cargo nextest run -p asap-logical-asap-optimization -E 'test(/asap_aware_cse::|reuse_candidates/)': 25 passed. This includes the new too_many_pane_or_reuse_candidates_name_the_count_found and the updated exact-message tests for the capability sharing and combined reuse caps.
• cargo nextest run -p asap-devtools --test stage_pipeline: 22 passed, including example1_document_is_valid_and_committed_fixture_is_current against the regenerated example.
• cargo nextest run -p asap-plan-selection -E 'test(cluster) | test(priced)': 6 passed.
• python3 -m unittest test_viewer test_server in tools/dag-viewer: 52 passed.
• cargo fmt --all -- --check and cargo clippy -p asap-executor -p asap-logical-asap-optimization -p asap-plan-selection -p asap-devtools --all-targets --all-features --locked -- -D warnings: clean.
• The full workspace suite was not run locally (per the repo's current practice); main's CI runs it after merge.

Removed unread field: the viewer's Stage 3 candidate row unit (stageCandidateRows in stages.js); nothing read it.

Limitations / follow-ups: pane_chain.rs folds the pane offset into the shift directly below the range only. That is correct, because the composed offset is additive and the @ anchor is now refused at any depth. It is noted here only in case stacked shifts become common.

• Stale Markdown (tracked in Update the docs after the G1–G11 consolidation groups land #677): tools/dag-viewer/README.md:84 still shows "unit": "cost_per_second"; it is now cost_unit.

[zzylol]

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

[zzylol]

Agreed, it was a special case. #758 replaces HydraRowWeight with a general RowWeight { Column(usize), Constant(NonZeroU32) }, and both keyed builds use it: the per-group keyed build (CMS or Count Sketch with heap) and the all-groups HydraCms build. The keyed per-group build now also takes a constant weight (1 for a row count, so a count-ranked heap binds at query time and precompute no longer projects a column of ones), an Int64 column and a nullable column. The Hydra-only i32::MAX bound stays in the all-groups constructor.

[zzylol]

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

[zzylol]

Re "SharedSummaryBuild just reading the name, I cannot tell what it is for": #758 renames the three executor summary builds by shape, so they read side by side. SummaryBuild → PerGroupSummaryBuild (one state per group, from a value column), KeyedSummaryBuild → PerGroupKeyedSummaryBuild (one keyed state per group whose items take a weight, e.g. CMS with heap), and SharedSummaryBuild → AllGroupsSummaryBuild (one state shared by all groups, emitted per group; HydraCms today). Their constructors are renamed to match, and each one has a doc comment. The serialized Operator variant names change with them.