fix: review nits after #745: cap message from its constant, Hydra constant weight, fallible updater, stacked shifts, viewer ranks

[zzylol]

Why

Review nits left after #745, decided on 2026-10-18:

• The five Pass 2 subset caps said "more than 20" in a hardcoded string, so the message would lie if a cap changed (feat(logical-asap-optimization): Example 2 — summary-capability rule and UnivMon frequency statistics #707 nit).
• validate_hydra_cms refused a constant weight such as Constant(2.0): only the unit count had a non-negative proof (feat(asap-physical-operators): execute HydraCms grouped counts and filtered aggregates #695/feat(dag-viewer): show the stage pipeline as three stacked lanes #698 nit).
• AccumulatorUpdater::update_keyed panicked (.expect) on a weighted-frequency sum past f64::MAX, and UnivMon's update_single on its counter overflow (fix(asap-physical-operators): seed each CountSketch row independently #706 nit).
• QueryWorkloadEntry derived Default, which only tests used.
• A per-entity summary looked through only one TimeShift below its range, so TimeRange(TimeShift(TimeShift(scan))) failed to compile (chore: fix the small nits collected from clean reviews #745 review, Low).
• The Stage Viewer ranked candidate rows by "has a cost" instead of by status (chore: fix the small nits collected from clean reviews #745 review, Low).
• An @ anchor in a counter window might be ignored (feat(planner): tumbling-pane window composition as a Pass 2 rule (Example 3) #713 nit). It is: see Counter window ignores the @ anchor (and its offset): rate(m[5m] @ 300) reads the window ending at the evaluation time #748.

What

• Cap errors are a new variant PerComputationCandidateError::TooManyCandidates { candidates, max }, whose message reads max from MAX_REUSE_CANDIDATES, MAX_PANE_CANDIDATES or MAX_CAPABILITY_SHARING_MEMBERS.
• A HydraCms build accepts Constant(w) with the new proof NonNegativeWeightProof::PositiveConstant when w is a positive integer within i32 (asap_sketchlib's Hydra adds i32 weights), and adds w per row. A non-positive, fractional or non-finite constant is refused at binding.
• AccumulatorUpdater::update_single and update_keyed return Result<(), String>. An overflow is returned to the caller instead of panicking, and update_single on a keyed updater is an error instead of a debug assertion.
• QueryWorkloadEntry has no Default. Tests name every field.
• A per-entity build and boundary_schema share one helper, below_time_shifts, which follows every stacked shift.
• The Stage Viewer ranks only selected and costlier rows.
• A test pins the @ behavior: rate(m[5m] @ 300) evaluated at 360 s reads the window ending at 360 s. The test is committed #[ignore]d, pointing at bug Counter window ignores the @ anchor (and its offset): rate(m[5m] @ 300) reads the window ending at the evaluation time #748, and this PR does not fix the bug.

Not in this PR:

• Heap pricing. Stage 3 still charges 24 B per heap entry. The real sizes now come from asap_sketchlib itself: feat(heap): estimated_memory_bytes for HHHeap and WeightedFrequency asap_sketchlib#233 adds HHHeap::estimated_memory_bytes and WeightedFrequency::estimated_memory_bytes. Once it merges, a follow-up PR moves the pin and has asap-plan-selection's summary_shape call them. asap-plan-selection already depends on asap_sketchlib (it calls ddsketch_indexable_bounds), so no new dependency is needed.
• CandidateCost keeps its name. It holds usage: ResourceUsage and also the priced cost_per_second, unit, source, and a NodeCost per node. It is a priced cost, not only usage, so renaming it to CandidateUsage would misname it.

How

• Cap message. Each cap site builds TooManyCandidates with a noun phrase and its constant. The error's #[error] formats "more than {max} {candidates} (2^n Pass 2 candidates)".
• Hydra constant weight.
  • validate_hydra_cms gains an arm (Constant(w), PositiveConstant), checked by hydra_constant_weight.
  • The physical planner passes the converted i32 to Operator::shared_summary_build as constant_weight, and the shared build adds it to each row when there is no weight column.
  • The executor operator Kind::SharedSummaryBuild stores it as an optional constant_weight field (#[serde(default = "unit_weight", skip_serializing_if = "is_unit_weight")]), so serialized operators without the field read as a count and a unit weight is not written. This adds an optional field to the serialized SharedSummaryBuild.
• Fallible updater. Each of the nine implementations returns Ok(()) or maps its kernel error. Callers in tests unwrap.
• Stacked shifts. below_time_shifts is generic over the node type, so it serves both the restored OperatorNode tree (per-entity build) and the physical node map (boundary_schema).

Key code interfaces added / modified

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

/// Why Stage 1 could not list a computation's or a Pass 2 rule's candidates.
/// The type stays; this PR adds one variant.
pub enum PerComputationCandidateError {
    /// An approximate measure's ε or δ is outside (0, 1): bad input.
    InvalidAccuracy { root: Option<String>, measure: Option<usize> },   // unchanged
    /// A choice names no listed candidate for a target: bad input.
    InvalidChoice,                                                      // unchanged
    /// A substitution this stage does not implement: an unsatisfiable request.
    Unsupported(&'static str),                                          // unchanged
    /// 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.
    TooManyCandidates {                                                 // new
        /// What was counted, e.g. "rollup candidates".
        candidates: &'static str,
        /// The cap that was exceeded (MAX_REUSE_CANDIDATES, MAX_PANE_CANDIDATES or
        /// MAX_CAPABILITY_SHARING_MEMBERS).
        max: usize,
    },
}

crates/types/src/ir/schema/sketch.rs

/// Why a summary update's weights are non-negative (what `WeightDomain::NonNegative`
/// carries). The type stays; this PR adds one variant.
pub enum NonNegativeWeightProof {
    /// Each row adds 1: a count.
    UnitCount,                       // unchanged
    /// A reset-corrected counter increase, or that increase over a positive duration.
    ResetAwareCounterDerivative,     // unchanged
    /// Raw samples of a metric declared a counter, or plain sums of them.
    CounterSamples,                  // unchanged
    /// A constant weight greater than 0, such as `Constant(2.0)`: every row adds the same
    /// positive amount, so no weight is negative.
    PositiveConstant,                // new
}

crates/types/src/workload/mod.rs

/// One query of the workload as the planner sees it. The type stays; it no longer
/// derives Default (only tests used it).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]   // was: also Default
pub struct QueryWorkloadEntry {
    /// The query text in the workload's language.
    pub query: Query,                       // unchanged
    /// Its accuracy target and latency bound.
    pub requirements: QueryRequirements,    // unchanged
    /// How far ahead it is known.
    pub predictability: Predictability,     // unchanged
    /// How often it runs.
    pub recurrence: QueryRecurrence,        // unchanged
}

crates/executor/src/summary_kernels/factory.rs

/// A summary state being built one input at a time; implemented by every kernel
/// updater. The trait stays; its two update methods become fallible.
pub trait AccumulatorUpdater: Send {
    /// Rejects a precompute input outside the kernel's domain before an update.
    fn validate_single_input(&self, value: f64) -> Result<(), String>;   // unchanged
    /// Updates from one typed row value.
    fn update_value(&mut self, value: &Value, timestamp_ms: i64) -> Result<(), String>;   // unchanged
    /// Feeds one (value, timestamp) pair. An error when the state cannot take the value
    /// (a counter or a sum past its range) or the updater is keyed; the state is then unchanged.
    fn update_single(&mut self, value: f64, timestamp_ms: i64) -> Result<(), String>;   // was -> ()
    /// Feeds one (key, value, timestamp) triple. An error when the state cannot take the
    /// value (a sum past f64::MAX); the state is then unchanged.
    fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, timestamp_ms: i64)
        -> Result<(), String>;                                                           // was -> ()
    /// Takes the built state out, resetting the updater.
    fn take_accumulator(&mut self) -> Box<dyn AggregateCore>;            // unchanged
    /// Reads the built state without resetting it.
    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore>;            // unchanged
    /// Consumes the updater and returns its state by move.
    fn into_accumulator(self: Box<Self>) -> Box<dyn AggregateCore>;      // unchanged
    /// Resets the state for reuse.
    fn reset(&mut self);                                                 // unchanged
    /// Whether the updater takes keyed updates.
    fn is_keyed(&self) -> bool;                                          // unchanged
    /// Estimated bytes the state holds.
    fn memory_usage_bytes(&self) -> usize;                               // unchanged
}

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

impl Operator {
    /// One HydraCms state shared by every group. Each row adds its `weight` column's
    /// value, or `constant_weight` without one (must be positive; 1 for a count).
    pub fn shared_summary_build(
        input: SchemaRef,
        family: SummaryFamilyType,
        item: usize,
        weight: Option<usize>,
        constant_weight: i32,   // new parameter
        groups: Vec<usize>,
    ) -> Result<Self, Error>;
}
/// A HydraCms constant weight as the i32 asap_sketchlib's Hydra adds: a positive integer
/// within i32; an error (bad input) otherwise. Read by validation and binding.
pub(crate) fn hydra_constant_weight(weight: f64) -> Result<i32, String>;   // new
/// The first node at or below `node` that is not a TimeShift, following each shift's child;
/// used by the per-entity build and boundary_schema.
pub(crate) fn below_time_shifts<'a, N, C: 'a>(
    node: &'a N,
    operator: impl Fn(&'a N) -> Option<&'a NonASAPOp<C>>,
    resolve: impl Fn(&'a C) -> Result<&'a N, Error>,
) -> Result<&'a N, Error>;   // new

tools/dag-viewer/stages.js

// stageCandidateRows(doc): one row per candidate; signature unchanged.
// `rank` is now given to rows whose status is 'selected' or 'costlier' (was: rows with a cost).

Usage, end to end: a deployment counting two requests per log line in a HydraCms state. It is checked by positive_constant_weight_adds_the_constant_per_row in crates/executor/tests/hydra_cms_execution.rs, which runs these steps at the head commit.

// 1. What the deployment hands the executor: a Hydra-grouped Count-Min SummaryAgg
//    (width 64, depth 3, a 3 x 64 shared grid) over rows (job, service, weight), grouped
//    by job, counting services. `update` is its SummaryUpdate: item = service, each row
//    adds the constant 2, proven non-negative by PositiveConstant. It is what step 2 binds.
let update = SummaryUpdate {
    item: Some(SummaryInputExpr::Column(ColumnRef::Named("service".into()))),
    weight: SummaryInputExpr::Constant(2.0),
    weight_domain: WeightDomain::NonNegative { proof: NonNegativeWeightProof::PositiveConstant },
};
// 2. Binding: compile_node -> validate_hydra_cms accepts (Constant(2.0), PositiveConstant)
//    via hydra_constant_weight(2.0) = 2, and summary_build passes constant_weight = 2 to
//    Operator::shared_summary_build. `build` is the bound operator step 3 runs.
let build = compile_node(&summary_agg(update, None), &[input_schema()])?;
// 3. Execution over the nine fixture rows; api has 2 "checkout" rows, batch and db 1 each.
//    `doubled` holds each job's "checkout" point count read back by SummaryEstimate.
let doubled = run(constant(2.0), None, vec![batch(&ROWS)]);
// 4. What the deployment gets back: twice the unit count's answer
//    (unit count {api: 2, batch: 1, db: 1}).
// Each job counts twice its unit count (api 4, batch 2, db 2), as the test asserts per job:
for job in ["api", "batch", "db"] {
    assert_eq!(doubled[job], 2.0 * unit[job], "{job}");
}
// A non-integer constant is bad input, refused at binding:
//    "HydraCms constant weight must be a positive integer within i32 ..., got 2.5".

And a deployment feeding a keyed state by hand (keyed_update_overflow_is_an_error_not_a_panic in crates/executor/tests/deployment.rs):

// `operator`: a plain Count Sketch (width 64, depth 3) updater keyed by host, from
// create_planner_accumulator; the deployment feeds it one weighted row at a time.
operator.update_keyed(&host(0), f64::MAX, 0)?;            // Ok: the cell holds f64::MAX
// A second f64::MAX would overflow the cell: today an Err the deployment handles
// (was a panic). The state keeps f64::MAX.
let error = operator.update_keyed(&host(0), f64::MAX, 1).unwrap_err();
assert!(error.contains("overflow"));

Before this PR

• More than 20 rollup candidates failed with "unsupported substitution …: more than 20 rollup candidates (2^n Pass 2 candidates)", a copy of the constant.
• A HydraCms SummaryAgg with Constant(2.0) failed binding with "HydraCms weight must be the unit count or a non-negative weight column".
• A keyed weighted-frequency or UnivMon overflow panicked inside the executor.
• Panes reading through two stacked shifts failed with "invalid DAG: per-entity summary requires a resolved source".
• A rejected row whose cost the document carried got a rank in the Stage Viewer.

After this PR

• The cap error reads "more than 20 rollup candidates (2^n Pass 2 candidates)", with 20 read from MAX_REUSE_CANDIDATES.
• Constant(2.0) builds and counts 2 per row.
• Overflows are errors returned to the caller.
• Stacked shifts compile and answer as the single build.
• Only valid rows are ranked.
• rate(m[5m] @ 300) is still wrong, tracked in Counter window ignores the @ anchor (and its offset): rate(m[5m] @ 300) reads the window ending at the evaluation time #748.

Verification

Each fix started with a failing test:

Test	Before the fix	Now
a_capability_sharing_candidate_past_20_members_is_an_error, reuse_candidates_past_20_together_are_an_error (now assert the variant and the message)	failed to compile (no variant)	pass
positive_constant_weight_adds_the_constant_per_row	failed to compile (no proof variant)	pass
keyed_update_overflow_is_an_error_not_a_panic	failed to compile (() has no unwrap)	pass
panes_over_stacked_time_shifts_match_single_build	failed: "per-entity summary requires a resolved source"	pass
test_status_follows_the_rejection_cause_not_the_cost (new rank assertions)	failed: rejected row ranked 19	pass
anchored_counter_window_ends_at_the_anchor	fails: [0.065, 0.065] vs [0.0667, 0.0667]	#[ignore] until #748

Tests run, each covering only the code it changes:

• cargo nextest run -p asap-logical-asap-optimization -E 'test(past_20) | test(pane) | test(rollup) | test(reuse)': 12 passed.
• cargo nextest run -p asap-executor --test hydra_cms_execution: 9 passed. -E 'test(hydra) | test(capability) | test(shared)': 17 passed.
• cargo nextest run -p asap-executor --test deployment --test promql_values --test weighted_topk_binding --test precompute_population --test precompute_candidates --test summary_projection: 37 passed. Lib -E 'test(kmv) | test(factory) | test(univmon) | test(planner_parameter)': 19 passed.
• cargo nextest run -p asap-integration-tests --test precompute_raw_samples: 3 passed. --test tumbling_pane_merge: 10 passed, 1 ignored. --test reuse_rules --test planner_layering_example1 --test promql_to_post_asap: 49 passed.
• cargo nextest run -p asap-frontend-sql --test pearson_corr: 7 passed. cargo test --doc -p asap-types workload: 7 passed.
• python3 -m unittest test_viewer (tools/dag-viewer): 37 run, OK (4 skipped).
• cargo fmt --all -- --check: clean. cargo clippy --all-targets --all-features --locked -p asap-types -p asap-logical-asap-optimization -p asap-executor -p asap-integration-tests -p asap-frontend-sql -p asap-frontend-promql -- -D warnings: clean.

No pinned example cost changes: this PR does not touch Stage 3 pricing.