Skip to content

feat(planner): Example 1 completion — identical-expression CSE, whole-expression top-k heaps, counter-sample weights, per-second plan-selection cost - #692

Merged
zzylol merged 15 commits into
mainfrom
stack/g4-example1-completion
Oct 14, 2026
Merged

zzylol merged 15 commits into
mainfrom
stack/g4-example1-completion

Conversation

@zzylol

@zzylol zzylol commented Oct 14, 2026 •

Copy link
Copy Markdown
Contributor

Why

#509's Example 1 could not run end to end on the planning stages:

  • Stage 1 had no Pass 2, so two queries over the same input could never share it.
  • Q2's heap sketches ranked an exact per-series sum instead of absorbing it, and their weights were read from the sample-value column, which is wrong for SQL keys.
  • Count-Min could never be admitted, because nothing proved counter samples non-negative.
  • Plan selection priced one evaluation of the whole workload, so queries with different recurrences could not be compared, and kept state had no price.

This PR completes Example 1 (consolidation group G4) under the user's decisions on these points.

What

  • Stage 1 Pass 2, identical-expression rule. One candidate per non-empty subset of a candidate's sharing groups that gives a distinct candidate. The independent candidate is kept. The variants are ordinary Stage 2/3 candidates, and nothing is capped in the library.
  • Whole-expression top-k heaps. A top-k over a per-item sum or count gets Count-Min + heap and CountSketch + heap candidates that read the inner aggregate's input and absorb it. They replace the heaps over the exact sum, so Example 1's Q2 has exactly three candidates.
  • Counter samples as an open requirement. Pass 1 records that a heap's weights are raw samples of metric M (or plain sums of them), without reading DataWorkload. Plan selection confirms it from the new DataWorkload.metric_types; unconfirmed counts as unmet.
  • Top-K heap weights (fix). A heap is weighted by the column the top-k ranks, so SQL with an Int64 key, or a key named value, plans correctly.
  • Per-second plan-selection cost.
    • Each query's recurrence is an input, and every term is a cost per second.
    • Memory is charged only as kept state (bytes × seconds); fix(planner): price state by its size; size KLL for δ #682's per-computation peak-memory charge is removed.
    • ResourceCalibration gains horizon_s and cost_per_retained_byte_second.
  • Devtool limit. stage_pipeline --max-combinations (default 10^4) counts Stage 1 candidates, Pass 2's included, while enumerating, and stops with an error past the limit. Exit 3 / server 422 are unchanged.

How

  • Pass 2 (asap_aware_cse). ir::cse interning identifies the sharing groups. Sharing a set of groups also shares every group it leaves as a single node. The distinct candidates are therefore the closed sets of that closure, and they are enumerated with Next Closure (at most one closure per step and group) instead of trying all 2^g subsets. They are emitted in the subsets' binary-counter order.
  • Stage 1 enumeration is a lazy iterator (stage1_logical_asap_optimization), and run_stage2_and_stage3 runs over what a caller enumerated. The devtool counts as it consumes the iterator; the library collects all candidates.
  • Whole-expression candidates (ComputationCandidate::WholeExpression) reuse the keyed-additive rule, which moves to realization.rs along with ranked_column. The outer partition keys are re-indexed onto the heap's input. Enumeration and combinations() skip choices that differ only in an absorbed target.
  • Requirements as symbols. The open weight requirement is WeightDomain::SamplesOfMetric { metric }. The sign of M's samples is the statistic values_non_negative[M], which gates a bound as a factor. Stage 3 binds it only to 1, when M is declared a counter, and leaves it unbound otherwise.
  • Cost (stage3_plan_selection module docs):
    • A NotMaterialized node is charged per consumer at the sum of their rates.
    • A MatAtQuery node is charged once at the combined rate of distinct intervals.
    • A MatAtIngestion node is charged per second of arriving rows.
    • One-off, scheduled and unknown queries are spread over horizon_s (3600 s).
    • Kept state is twice its bytes at ingestion and once at query time.
    • Totals within 1e-9 relative tie, and a tie keeps the earlier candidate.

Key code interfaces added / modified

crates/asap-aware-mapping/src/asap_aware_cse.rs (new), lib.rs

/// Stage 1, Pass 2: ASAP-aware CSE (the identical-expression rule only).
pub mod asap_aware_cse; // new: Pass 2 identical-expression rule

/// Two or more identical nodes that are legal to share, in children-first walk order.
pub struct SharingGroup { pub nodes: Vec<Rc<OperatorNode>> }   // from ir::cse interning
/// One Pass 2 candidate: the input roots with the listed sharing groups each made one node.
pub struct IdenticalExpressionVariant {
    /// Indices into the `sharing_groups` result of the groups this variant shares.
    pub shared_groups: Vec<usize>,
    /// The candidate's query roots with those groups shared.
    pub roots: Vec<QueryRoot>,
}

/// The sharing groups of a whole-workload candidate's roots, children before parents.
pub fn sharing_groups(roots: &[QueryRoot]) -> Vec<SharingGroup>;
/// One variant per distinct closed subset of groups (Next Closure).
/// Empty when nothing is identical; the caller keeps the unshared candidate beside them.
pub fn identical_expression_variants(roots: &[QueryRoot], groups: &[SharingGroup]) -> Vec<IdenticalExpressionVariant>;

crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs, pass/mod.rs

/// One Stage 1 candidate in enumeration order, before Stage 2 numbers it.
pub struct Stage1Candidate {                       // new: one enumerated Stage 1 candidate
    /// Human-readable name, used only for display.
    pub display_name: String,
    /// The Pass 1 choice per target; a Pass 2 candidate has its Pass 1 candidate's.
    pub choice: Vec<usize>,
    /// The composed query roots, or why the choice did not compose.
    pub roots: Result<Vec<QueryRoot>, PerComputationCandidateError>,
    /// For a Pass 2 candidate: its Pass 1 candidate's enumeration position and the
    /// shared groups (operator kinds); `None` for a Pass 1 candidate.
    pub pass2_of: Option<(usize, Vec<String>)>,
}
/// A candidate Pass 2 added: `id` is `from_logical` with each group in `shared` made one node.
pub struct IdenticalExpressionCandidate { pub id: String, pub from_logical: String, pub shared: Vec<String> } // new

/// The output of Stages 1 to 3 for one workload. The type stays; it gains two fields.
pub struct Stage1ToStage3Run<Id> {
    /// Stage 1, Pass 1: the per-computation candidates, with the roots they read.
    pub stage1_pass1_per_computation: PerComputationCandidates<Id>,          // unchanged
    /// Stage 1, Pass 1: how many whole-workload candidates compose; Pass 2 adds to them.
    pub combinations: usize,                                                 // unchanged
    /// Stage 1, Pass 2: the candidates the identical-expression rule added, in enumeration order.
    pub stage1_pass2_identical_expression: Vec<IdenticalExpressionCandidate>, // new
    /// Stage 1: every whole-workload candidate, each Pass 1 one followed by its Pass 2 ones.
    pub stage1_logical_asap_optimization: Vec<LogicalCandidate>,             // unchanged
    /// Stage 2: one physical candidate `P<n>` per implemented `L<n>`.
    pub stage2_physical_asap_optimization: Vec<PhysicalCandidate>,           // unchanged
    /// Stage 2: the Stage 1 candidates it could not implement, with reasons.
    pub stage2_pruned: Vec<PrunedCandidate>,                                 // unchanged
    /// Stage 3: plan selection over the Stage 2 candidates, or why it failed.
    pub stage3_plan_selection: Result<Selection, SelectionError>,            // unchanged
    /// Stage 3: the selected candidate's logical roots with confirmed weight requirements;
    /// `None` when selection failed.
    pub stage3_selected_logical: Option<Vec<QueryRoot>>,                     // new
}
impl<Id> Stage1ToStage3Run<Id> {
    /// The selected logical plan, with the weight requirements Stage 3 confirmed made proofs.
    pub fn selected_logical(&self) -> Option<&[QueryRoot]>; // now the confirmed plan
}

/// Lazy, uncapped Stage 1 enumeration; a caller that limits candidates stops consuming it.
// new: lazy Stage 1 enumeration, Pass 1 choices each followed by their Pass 2 variants
pub fn stage1_logical_asap_optimization<Id: Clone>(
    per_computation: &PerComputationCandidates<Id>,
) -> impl Iterator<Item = Stage1Candidate> + '_;
/// Runs Stage 2 and Stage 3 over `stage1` (all candidates, or the ones a caller
/// enumerated); candidate `n` becomes `L<n>`, implemented as `P<n>`.
// new: Stage 2/3 over caller-enumerated candidates
pub fn run_stage2_and_stage3<Id: Clone + std::fmt::Debug>(
    per_computation: PerComputationCandidates<Id>,
    stage1: Vec<Stage1Candidate>,
    targets: &[Option<AccuracyTarget>],
    recurrence: &[QueryRecurrence],
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Stage1ToStage3Run<Id>;
/// Runs Stages 1 to 3 over the frontends' roots; `recurrence[i]` is how often root `i`
/// is evaluated. Only a Pass 1 failure is an `Err`.
pub fn run_stage1_to_stage3<Id: Clone + std::fmt::Debug>(
    roots: Vec<(Id, QueryRoot)>,
    targets: &[Option<AccuracyTarget>],
    recurrence: &[QueryRecurrence],                // new parameter
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Result<Stage1ToStage3Run<Id>, PerComputationCandidateError>;
/// As `run_stage1_to_stage3`, from an existing Pass 1 inventory, so a caller can
/// inspect it before enumerating every Stage 1 candidate.
pub fn run_stage1_to_stage3_from_pass1<Id: Clone + std::fmt::Debug>(
    per_computation: PerComputationCandidates<Id>,
    targets: &[Option<AccuracyTarget>],
    recurrence: &[QueryRecurrence],                // new parameter
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Stage1ToStage3Run<Id>;

crates/asap-aware-mapping/src/per_computation_candidates.rs

/// One Pass 1 candidate for a computation (one measure of an aggregate).
/// The type stays; it gains `WholeExpression`.
pub enum ComputationCandidate {
    /// Plain exact execution of the original sub-DAG, when no exact accumulator form applies.
    Exact,                                                         // unchanged
    /// An exact mergeable accumulator state.
    ExactAccumulator { kind: ExactKind, params: ExactParams },     // unchanged
    /// An unsized summary family, sized later for the measure's accuracy.
    Summary(SketchAlgorithm),                                      // unchanged
    /// `avg` rewritten as exact `sum` and `count` accumulator states.
    AvgAsSumCount,                                                 // unchanged
    /// Top-k as one unsized heap summary over the input of the per-item sum/count beneath;
    /// replaces the same-family heap over the exact per-item result.
    WholeExpression(SketchAlgorithm), // new: whole-expression Count-Min/CountSketch + heap absorbing the inner sum/count
}

/// One measure of an aggregate with at least one realization besides exact execution.
/// The type stays; it gains `absorbs` and `samples_of`.
pub struct ComputationTarget {
    /// The aggregate node this target realizes.
    pub target: Rc<OperatorNode>,     // unchanged
    /// Index into the aggregate's measures.
    pub measure: usize,               // unchanged
    /// The candidates for this measure, exact first.
    pub choices: ComputationChoices,  // unchanged
    /// The measure's own filter; only rows that pass it update the state (`SummaryAgg.filter`).
    pub filter: Option<Predicate>,    // unchanged
    /// The aggregate's own `having`, then one entry per consumer `Filter` directly above it.
    pub having: Vec<HavingEntry>,     // unchanged
    /// The target directly beneath that this target's `WholeExpression` candidates
    /// absorb (not computed when chosen); `None` otherwise.
    pub absorbs: Option<usize>,       // new: the target a WholeExpression candidate absorbs
    /// The metric whose raw samples (or plain sums) are every value the aggregate reads;
    /// `None` otherwise. Heap weights from it carry `WeightDomain::SamplesOfMetric`.
    pub samples_of: Option<String>,   // new: the metric whose raw samples feed the aggregate
}

impl<Id> PerComputationCandidates<Id> {
    /// Whether `choice` picks a whole-expression candidate and a non-first candidate for
    /// the target it absorbs (a duplicate of the first), so enumeration skips it.
    // new; combinations() now excludes choices differing only in an absorbed target
    pub fn absorbed_choice_varies(&self, choice: &[usize]) -> bool;
}

crates/asap-aware-mapping/src/stage3_plan_selection.rs, analytical_cost.rs

/// Unit of every Stage 3 cost: cost per second of wall time (one unit is one
/// CPU-millisecond under `illustrative_calibration`).
pub const COST_UNIT: &str = "cost_per_second";                              // was "cpu_ms_per_workload_evaluation"
/// Label naming the cost model and statistics the costs come from.
pub const COST_SOURCE: &str = "analytical-cost-v2 (illustrative statistics)"; // was v1

/// Stage 3: rejects candidates that miss a query's accuracy target, prices the rest
/// per second and selects the cheapest (first on ties).
pub fn select_plan(
    cands: &[PhysicalCandidate],
    targets: &[Option<AccuracyTarget>],
    recurrence: &[QueryRecurrence],   // new: prices per second from each query's recurrence
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Result<Selection, SelectionError>;
/// Coefficients used until a deployment is calibrated: 1 ns CPU per op, 0.1 ns per
/// scanned byte, 1 GB kept for a second = 125 CPU-ms, one-off work spread over an hour.
pub fn illustrative_calibration() -> ResourceCalibration;                              // new
/// Answers an open `SamplesOfMetric` requirement from `data`'s declared metric types:
/// proven for a declared counter, unknown otherwise.
pub fn confirmed_weights(domain: &WeightDomain, data: &DataWorkload) -> WeightDomain;   // new
/// `roots` with every open weight requirement replaced by its answer: the form of a
/// selected plan the executor runs.
pub fn with_confirmed_weights(roots: &[QueryRoot], data: &DataWorkload) -> Vec<QueryRoot>; // new

/// Stage 3's result. The type stays; it gains `plan`.
pub struct Selection {
    /// The selected physical candidate's id, e.g. "P1".
    pub selected: String,                          // unchanged
    /// The selected candidate's roots with every confirmed weight requirement made a proof: the plan to run.
    pub plan: Vec<QueryRoot>,                      // new
    /// Each valid candidate's cost, by id.
    pub costs: BTreeMap<String, CandidateCost>,    // unchanged
    /// Every other candidate: invalid with the reason, or valid but costlier/tied.
    pub rejected: Vec<Rejection>,                  // unchanged
}
/// One node's price. The type stays; it gains `per_second`.
pub struct NodeCost {
    /// Price per second: one computation's price × `multiplier` × `per_second`.
    pub cost: f64,                                 // unchanged (now per second)
    /// The consumers it is computed for; 1 when materialized.
    pub multiplier: u64,                           // unchanged
    /// Computations per second, from the readers' recurrences.
    pub per_second: f64,                           // new
    /// A human-readable note, e.g. "scan 3000000 samples".
    pub detail: String,                            // unchanged
}
/// Why plan selection produced no selection. The type stays.
pub enum SelectionError {
    /// A candidate's root count differs from the targets or recurrences given;
    /// a caller bug (mismatched inputs).
    TargetCount { candidate: String, roots: usize, targets: usize, recurrences: usize },  // recurrences new
    /// Pricing one node failed; an internal bug or an uncostable operator.
    Cost { candidate: String, node: PhysicalASAPNodeId, error: AnalyticalCostError },     // unchanged
    /// Every candidate was rejected; an unsatisfiable request, with each rejection.
    NoValidCandidate(Vec<Rejection>),                                                     // unchanged
}

// analytical_cost.rs
/// Conversion from physical dimensions to one deployment-specific cost objective.
/// The type stays; it gains two fields.
pub struct ResourceCalibration {
    /// Price of one CPU operation.
    pub cost_per_cpu_op: f64,                 // unchanged
    /// Price of one scanned byte.
    pub cost_per_scan_byte: f64,              // unchanged
    /// Price of one byte held by a computation.
    pub cost_per_retained_byte: f64,          // unchanged
    /// Price of state kept across evaluations, per byte per second; default 0.
    #[serde(default)]
    pub cost_per_retained_byte_second: f64,   // new: price of state kept across evaluations
    /// Seconds over which one-off, scheduled and unknown queries are spread;
    /// default `DEFAULT_HORIZON_S`.
    #[serde(default = "default_horizon_s")]
    pub horizon_s: f64,                       // new: spread for one-off, scheduled and unknown queries
    /// Names the calibration in cost sources.
    pub version: String,                      // unchanged
}
/// Default `horizon_s`: one hour.
pub const DEFAULT_HORIZON_S: f64 = 3_600.0;   // new

crates/asap-aware-mapping/src/accuracy/evidence.rs, composition.rs

/// Statistics a propagation rule may consult; every field defaults to unknown.
/// The type stays; it gains `sign_statistic`.
pub struct PropagationStats {
    /// Certified ranges for a division's true numerator and denominator.
    pub division_operand_domains: Option<[OperandDomain; 2]>,   // unchanged
    /// Provenance of the supplied evidence.
    pub evidence_provenance: Vec<GuaranteeSource>,              // unchanged
    /// Whether every input value is non-negative.
    pub values_non_negative: Option<bool>,                      // unchanged
    /// Names the statistic an unknown sign is gated on when the values are one metric's
    /// raw samples or sums; plan selection binds it to 1 for a declared counter. Default `None`.
    pub sign_statistic: Option<String>, // new: names the values_non_negative[M] statistic gating a relative bound
    /// Input rows an exact aggregation folds.
    pub input_row_count: Option<u64>,                           // unchanged
    /// The statistic an unknown row count is named by (default `input_row_count`).
    pub row_count_statistic: Option<&'static str>,              // unchanged
    /// Key-frequency distribution evidence.
    pub data_distribution: Option<DataDistribution>,            // unchanged
    /// Lower confidence bound of the k-th selected top-k item.
    pub topk_selected_lower_bound: Option<f64>,                 // unchanged
    /// Greatest upper confidence bound among excluded top-k items.
    pub topk_excluded_upper_bound: Option<f64>,                 // unchanged
    /// Union-bound failure probability of the top-k margin intervals.
    pub topk_interval_failure_probability: Option<f64>,         // unchanged
    /// Hydra shared-grid collision error.
    pub hydra_shared_grid_collision_bound: Option<f64>,         // unchanged
    /// Failure probability of the Hydra shared-grid term.
    pub hydra_shared_grid_failure_probability: Option<f64>,     // unchanged
}
/// The `values_non_negative[metric]` statistic name for samples of `metric`.
pub(crate) fn samples_sign_statistic(metric: &str) -> String;                    // new
/// The metric a `samples_sign_statistic` name refers to; `None` if it is not one.
pub(crate) fn parse_samples_sign_statistic(statistic: &str) -> Option<&str>;     // new

crates/types/src/post_asap/sketch.rs, workload.rs, ir/schema_support.rs

/// What is known about the sign of a summary update's weights. The type stays; it gains `SamplesOfMetric`.
pub enum WeightDomain {
    /// Nothing proven; the weights may be negative. The default.
    UnknownOrSigned,                                // unchanged
    /// The weights are proven non-negative, by `proof`.
    NonNegative { proof: NonNegativeWeightProof },  // unchanged
    /// Open requirement: weights are raw samples (or plain sums) of `metric`; plan
    /// selection confirms it, and unconfirmed it counts as not proven.
    SamplesOfMetric { metric: String },   // new: open requirement recorded by Pass 1
}
impl WeightDomain {
    /// Answers an open `SamplesOfMetric`: `CounterSamples` proof if `is_counter`, else
    /// `UnknownOrSigned`; any other domain is returned unchanged.
    pub fn confirmed(&self, is_counter: impl Fn(&str) -> bool) -> WeightDomain; // new
}
/// Why weights are non-negative. The type stays; it gains `CounterSamples`.
pub enum NonNegativeWeightProof {
    /// Every weight is 1 (a count).
    UnitCount,                     // unchanged
    /// The weights are a reset-aware counter derivative (`rate`, `increase`).
    ResetAwareCounterDerivative,   // unchanged
    /// The weights are raw samples of a metric the data workload declares a counter.
    CounterSamples,                // new
}

/// The data workload: statistics and declarations about the data being queried.
/// The type stays; it gains `metric_types`.
pub struct DataWorkload {
    /// Whether data keeps arriving or is at rest.
    pub arrival: DataArrival,                                          // unchanged
    /// Cadence at which each PromQL source supplies a sample.
    pub data_ingestion_interval: Evidence<DurationMs>,                 // unchanged
    /// Total ingested volume.
    pub ingestion_volume: Evidence<u64>,                               // unchanged
    /// Rows per second arriving.
    pub ingestion_rate: Evidence<Rate>,                                // unchanged
    /// Input rows (series) per evaluation.
    pub input_cardinality: Evidence<u64>,                              // unchanged
    /// Key-frequency distribution.
    pub distribution: Evidence<DataDistribution>,                      // unchanged
    /// Per-source L1/L2 norm bounds for Count-Min / Count Sketch errors.
    pub stream_norms: Evidence<BTreeMap<String, StreamNorms>>,         // unchanged
    /// Per-source bound on distinct items per evaluation (classic HLL's failure probability).
    pub max_distinct_per_evaluation: Evidence<BTreeMap<String, u64>>,  // unchanged
    /// Declared Prometheus type of each metric by name; never inferred from a name.
    /// Default empty (an undeclared metric's samples may take any value).
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub metric_types: BTreeMap<String, MetricType>, // new: optional in workload JSON
}
/// The Prometheus type of a metric's samples: `Counter` (only increases or resets to
/// zero, so every sample is non-negative) or `Gauge` (may take any value).
#[serde(rename_all = "snake_case")]
pub enum MetricType { Counter, Gauge } // new

/// Resolves a PromQL root to rows carrying the series identity before candidate search.
// now adds a (series identity, timestamp) unique key so Pass 2 may share the node
pub fn with_promql_series_identity(root: &Rc<OperatorNode>) -> Result<Rc<OperatorNode>, String>;

crates/devtools/src/planning_stages.rs, bin/stage_pipeline.rs

/// Runs the planning stages over roots lowered from any query language; query `i` has
/// target `targets[i]` and recurrence `recurrence[i]`.
pub fn run_planning_stages_over(
    roots: Vec<QueryRoot>,
    targets: &[AccuracyTarget],
    recurrence: &[QueryRecurrence],   // new parameter
    data: &DataWorkload,
    max_combinations: Option<usize>,  // now a Stage 1 candidate limit, Pass 2 included
) -> Result<PlanningStagesRun, PlanningStagesError>;

/// Why a planning-stages run produced nothing. The type stays; one variant is replaced.
pub enum PlanningStagesError {
    /// Pass 1 failed: there are no candidates, only the frontends' roots.
    Pass1(String),                       // unchanged
    /// Stage 1 enumerated more candidates than the caller's limit; enumeration stopped and
    /// nothing was planned. An unsatisfiable request under that limit, not a bug.
    TooManyCandidates { limit: usize }, // was TooManyCombinations { combinations, limit }
    /// The frontends' roots or a composed candidate failed validation; an internal bug.
    Validation(String),                  // unchanged
}
// CLI flag: the most Stage 1 candidates a run enumerates; past it the run writes no document.
// stage_pipeline --max-combinations: now a Stage 1 candidate limit (default 10^4); exit 3 unchanged

Usage, end to end (checked with a scratch test over Example 1 at 80fc72e1): the library path a caller such as stage_pipeline takes, from the user's workload to the selected plan.

// What the user writes: Example 1's two panels, both every 10 s over the last minute.
//   Q1 "sum by (job) (rate(http_requests_total[1m]))", exact
//   Q2 "topk by (job) (10, sum_over_time(http_requests_total[1m]))", ε = 0.01, δ = 0.001
// plus data_workload.metric_types = { "http_requests_total": Counter } (new in this PR).
// `workload` is read below for the queries, their targets and recurrences, and the data facts.
let workload = example1();

// `roots`: the PromQL frontend's DAG, one root per query (Stage 0); the input to Pass 1.
let roots = lower_promql_roots(&workload)?;
// `targets`: each query's accuracy target [Exact, EpsilonDelta { 0.01, 0.001 }]; Stage 3 checks guarantees against it.
let targets: Vec<_> = workload.query_workload.entries().map(|e| Some(e.requirements.accuracy.target())).collect();
// `recurrence`: each query's demand, [FixedInterval(10 s); 2]; Stage 3 turns it into cost per second (new).
let recurrence: Vec<_> = workload.query_workload.entries().map(|e| e.recurrence).collect();
// `data`: the data workload; Stage 3 prices rows from it and confirms counter samples from `metric_types`.
let data = workload.data_workload.clone().unwrap_or_default();

// `per_computation`: Stage 1 / Pass 1's candidates per computation. Q2's top-k offers exact,
// whole-expression CMS+heap and whole-expression CountSketch+heap (new: the heaps absorb the inner
// per-job sum and record their weights as WeightDomain::SamplesOfMetric { metric: "http_requests_total" }).
// 3 whole-workload combinations; it is the input to the Stage 1 enumeration.
let per_computation = generate_per_computation_candidates(roots.iter().cloned().enumerate().collect())?;

// `stage1`: Stage 1's lazy enumeration, collected: each Pass 1 combination followed by its Pass 2
// identical-expression candidates (new), 3 × (independent, shared Scan #1, shared Scan #1 + TimeRange #2)
// = L1..L9. A caller with a limit (stage_pipeline --max-combinations) stops consuming the iterator there.
let stage1: Vec<Stage1Candidate> = stage1_logical_asap_optimization(&per_computation).collect();

// `run`: Stage 2 implements L<n> as P<n>; Stage 3 binds the counter requirement from `data`,
// checks every target, prices every valid candidate in cost per second, and selects the cheapest.
let run = run_stage2_and_stage3(per_computation, stage1, &targets, &recurrence, &data, PlanningModels::builtin());

// `selection`: what the user gets, the selected plan and every other candidate's verdict.
let selection = run.stage3_plan_selection.as_ref()?;
// selection.selected == "P1"   all exact, 7.94021 cost_per_second
// P2, P3 (shared scan, shared scan and range): valid, "tied with the selected candidate P1 at 7.940 cost_per_second; lost on enumeration order"
// P4–P6 (CMS+heap): weights confirmed non-negative from Counter, rejected only because "q2: the output has no guarantee, …"
//   (without the Counter declaration: "q2: CmsWithHeap needs non-negative update weights, and these are not proven non-negative …")
// P7–P9 (CountSketch+heap): "q2: the output has no guarantee, …"

// `plan`: the selected candidate's two logical roots with every weight requirement Stage 3 confirmed
// turned into a proof; this is what the executor is handed, so a selected Count-Min heap is accepted.
let plan = run.selected_logical();

Before this PR

  • Example 1. Stage 1 had 3 candidates and no sharing. Q2's heaps ranked an exact per-series sum, and Count-Min was always rejected on its weights.
  • SQL top-k. With an Int64 key the heap candidates failed to compose; with a key named value the heap was weighted by the key.
  • Costs were per workload evaluation, ignoring recurrence. A computation was charged for the memory it held while running, and kept state was not priced per second.
  • Devtool limit. stage_pipeline refused workloads by their Pass 1 combination count only.

After this PR

  • Example 1 has 9 Stage 1 candidates: 3 Pass 1 × (independent, shared scan, shared scan and range).
    • Q2 offers exact, whole-expression Count-Min + heap and whole-expression CountSketch + heap.
    • With http_requests_total declared a counter, Count-Min is rejected only because a top-k output has no guarantee; without the declaration it is rejected on its weights.
    • The shared-input plans tie with the independent one, which is kept.
  • Example 3a has 972 candidates.
  • Library callers of run_stage1_to_stage3 get Selection::plan and selected_logical() with confirmed weights, so the executor accepts a selected Count-Min heap.
  • Evidence: the selected plan and total per example, measured at the previous head of this PR and at this head.
Example Before (previous head) After
1 P1, exact Q2: 8.105 P1, exact Q2: 7.940
2 P1, all exact: 13.440 P1, all exact: 8.640 (tied with its shared-input plans)
3a P433, KLL for Q1–Q2, exact for Q3–Q5: 21725.736 P485, KLL for all five: 21724.827; all-exact costs about 16% more
3b P1, exact: 2.720 P2, KLL: 2.100, tied with DDSketch; exact 2.400

Totals are cost per second under the illustrative calibration. Screenshots: not applicable.

Verification

  • Failing test first for each fix and decision:
    • SQL Int64 key and key named value.
    • Kept-state-only memory.
    • Sign binding for declared counters, and the gate opening only at 1.
    • Candidate limit including Pass 2.
    • The confirmed plan reaching every consumer.
    • An open requirement not certifying Count-Min.
    • 20 nested sharing groups enumerated quickly.
  • Brute-force equivalence of the Pass 2 enumeration on small cases, and a label/roots agreement test.
  • Example acceptance tests:
    • Example 1: the Pass 2 tests are un-ignored, the accumulator count and Count-Min reasons are updated, and a tie-tolerance test is added.
    • Example 3: the Pass 2 test is un-ignored.
    • The whole-expression heaps are now checked: the raw-sample precompute test requires both heap families, and the numeric-regression test is un-ignored.
  • Fixtures and docs. The viewer fixtures are regenerated. The Example 3 acceptance spec and the stage viewer guide state the new selections.
  • Full CI gate on the remote runner at d542c30: fmt and clippy --all-features --locked -D warnings pass, and nextest over the workspace passes 1650 tests (46 skipped).

Limitations and follow-ups

Supersedes #588, #589, #640, #593 and #594, which are left open.

zzylol added 15 commits October 14, 2026 22:17
…ked value

Pass 1's Top-K sketch update weighted every heap by the sample value. Over
SQL rows that column is the one named `value`, or else the only numeric
column: an Int64 key made the heap candidates fail to compose, and a key
named `value` would have weighted the heap by the key.

The update now reads the column the top-k ranks (the aggregate's measure
over grouped rows, the sample value over per-series rows), the same column
exact top-k sorts by. `ranked_column` moves from Stage 2 into the shared
realization helpers so both stages use it.
…Stage 3 candidates

Stage 1 Pass 2 now applies the identical-expression rule of Algorithm 2.
For each composed Pass 1 candidate it finds the sharing groups (two or
more nodes with identical input and computation semantics that are legal
to share) and adds one candidate per non-empty subset of the groups,
keeping the independent candidate. A subset that leaves out a group that
the other shared groups already merge gives the same candidate as the
larger subset, so it is listed once. Nothing is capped. The variants are
ordinary candidates for Stage 2 and Stage 3.

PromQL rows that carry the full series identity now declare
(identity, timestamp) as a unique key, which makes scans and range
selectors legal to share.

Plan selection treats totals that differ only by floating-point rounding
as a tie and keeps the earlier candidate. A shared NotMaterialized input
is charged once per consumer, so it ties with the independent plan.

Example 1 gets 9 logical candidates (3 Pass 1 x (1 + 2 Pass 2)) and
Example 3a gets 972. stage_pipeline's --max-combinations still counts
Pass 1 combinations; its document lists every Stage 1 candidate. The
Example 1 and scalar-queries viewer fixtures are regenerated.
…e into the realization helpers

Pass 1's whole-expression heap reuses the legacy search's keyed-additive
rule, so the rule, its result types and schema_column_ref move from
replacement.rs into realization.rs, which both use. Pure move.
…nter-sample weights

Pass 1 gains a whole-expression candidate kind. A top-k over a per-item
sum or count that is itself a target (one consumer, no filter or HAVING,
realizable by the keyed-additive rule) has its Count-Min and CountSketch
heap candidates replaced by whole-expression ones: one heap over the
inner aggregate's input, keyed by the ranked item and weighted by the
summed value. It absorbs the inner target, which is then not computed.
The outer top-k's partition keys are moved onto the input the heap reads.
Example 1's Q2 gets exactly three candidates: exact, Count-Min + heap and
CountSketch + heap. Enumeration skips choices that differ only in an
absorbed target, and PerComputationCandidates::combinations counts the
same set.

Count-Min needs non-negative weights. Pass 1 does not read the data
workload: when a heap's weights are raw samples of metric M, or plain sums
of them, it records the open requirement WeightDomain::SamplesOfMetric{M}.
Rate, other arithmetic and other functions end it, and no type is inferred
from a `_total` name. DataWorkload gains metric_types (declared Prometheus
types). Plan selection admits Count-Min over such weights only when M is
declared a counter; an unconfirmed requirement counts as unmet. The
planner pipeline's selected plan carries the confirmed requirement as the
proof NonNegativeWeightProof::CounterSamples, since the executor reads no
data workload.

Example 1 declares http_requests_total a counter, so its Count-Min heap
is now rejected only because a top-k output has no guarantee; without the
declaration it is rejected on its weights. Raw-sample precompute now
checks the whole-expression heaps. The Example 1 viewer fixture is
regenerated.
Plan selection now prices a candidate per second, so queries with
different recurrences add up, and takes each query's recurrence:

- A NotMaterialized node is charged once per consumer at the sum of their
  rates (a shared NotMaterialized input still costs what reading it
  separately does).
- A MatAtQuery node is computed once at the combined rate of the queries
  reaching it; queries at the same interval run together.
- A MatAtIngestion node is priced over one second of arriving rows, once
  per second.
- One-off, scheduled and unknown queries are spread over horizon_s.

ResourceCalibration gains horizon_s (default 3600 s) and
cost_per_retained_byte_second, the price of state kept between
computations. Every materialized node's kept state is charged per byte per
second: twice its output at ingestion time (the window being built and
the completed one), once at query time. Memory held only during a query
gets no per-second charge. select_plan, run_stage1_to_stage3 and
run_stage1_to_stage3_from_pass1 take the recurrences; the planner
pipeline and stage_pipeline read them from the workload. The cost model
is described in the stage3_plan_selection module docs. The viewer
fixtures are regenerated.
Plan selection no longer charges a computation for the memory it holds
while it runs (#682's per-computation peak-memory term, including the
growing exact state of a query-time aggregate). Memory is charged only as
state a materialized node keeps between computations: kept bytes times
cost_per_retained_byte_second. The kept state is the node's output, or
the values an exact aggregate keeps per group when that is more, so a kept
exact quantile or distinct count is still sized by what it keeps. Peak
capacity belongs to cost model 2 (#683). Every term is now a per-second
cost; the per-computation memory coefficient is ignored.

The tests that compared exact state against sketches now keep the state
at query time, where its size is charged.
An exact sum or mean of relative-error values keeps the relative bound
only over non-negative values. When the folded values are raw samples of
one metric, or plain sums of them (an exact sum accumulator, a renaming,
or a heap estimate whose weights carry the SamplesOfMetric requirement),
Stage 1 now names the unknown sign by that metric,
values_non_negative[M], and gates the bound on it. Plan selection binds it
to 1 when the data workload declares M a counter; a gauge, an undeclared
metric, rate, other arithmetic or a name such as _total leave it unbound,
so the target stays unmet. Other folds keep the bare values_non_negative,
which nothing binds.
…umerating

stage_pipeline's --max-combinations (default 10^4) now counts Stage 1
candidates as they are enumerated, Pass 2's identical-expression
candidates included, and stops with an error as soon as the count passes
the limit; it never truncates. Exit status 3 (the viewer server's 422)
is unchanged. The library stays uncapped: Stage 1 is exposed as a lazy
iterator (stage1_logical_asap_optimization) and Stage 2 and Stage 3 run
over the candidates a caller enumerated (run_stage2_and_stage3);
run_stage1_to_stage3_from_pass1 enumerates them all. planning_stages
takes the limit as an Option.
… directly

The identical-expression rule tried all 2^g subsets of the sharing groups
and dropped those that repeat a candidate. With nested groups most
subsets repeat (14 groups: 16 384 subsets for 130 candidates). Sharing a
set of groups also shares the groups it leaves as one node; that closure
maps each subset to its candidate, and the distinct candidates are its
closed sets. They are now enumerated directly (Next Closure: one closure
per step and group), then put in the same binary-counter order, so the
output is unchanged. Twenty nested groups give their 20 candidates
quickly; a brute-force test checks the same candidates on small cases.

Sharing groups are computed once per candidate, and a Pass 2 candidate
names its shared groups by kind and number (Scan #1, TimeRange #2), so
repeated kinds stay distinguishable. The viewer fixtures are regenerated.
…rmed weights

The counter-sample requirement Stage 3 confirms was made a proof only in
the planner pass, so run_stage1_to_stage3's selected_logical and the
selected candidate's roots still carried the open requirement, which the
executor rejects. Stage 3 now puts the plan on its output:
Selection::plan is the selected candidate's roots with confirmed weights,
and Stage1ToStage3Run::selected_logical returns the selected logical
roots the same way. The pass reads Selection::plan.
…ertify Count-Min

estimate_has_bound treated any weight domain but UnknownOrSigned as
proven, so a Count-Min estimate over weights with the open
SamplesOfMetric requirement got its bound as if the weights were proven
non-negative. Only NonNegative now counts. Such an estimate instead gets
its local bound gated on the metric's sign, values_non_negative[M], which
plan selection binds to 1 when M is declared a counter and leaves unbound
otherwise.
… to rounding

Shows that the tie tolerance is needed: the shared-input candidates'
totals differ from the independent candidate's in the last bits, and
plan selection reports them as tied and keeps the independent one. Also
drops stale wording in the acceptance helpers.
Applying a candidate's label (the groups it shares) gives the same roots
the direct enumeration returned, sharing exactly those groups.
The sign of one metric's samples gates a relative bound as a factor; a
binder supplying 0 would have read as a perfect bound. Plan selection now
binds a sign statistic only to 1 (proven) and leaves any other value
unbound.
The Example 3 acceptance spec and the stage viewer guide now give 3a's
972 candidates and its all-KLL selection, 3b's KLL selection tied with
DDSketch, the per-second totals, and the shared-input plans that tie with
the independent one.
@zzylol
zzylol force-pushed the stack/g4-example1-completion branch from 4009093 to d542c30 Compare October 14, 2026 23:16
@zzylol zzylol changed the title feat(logical-asap-optimization): Example 1 completion — identical-expression CSE, whole-expression top-k heaps, per-second plan-selection cost feat(planner): Example 1 completion — identical-expression CSE, whole-expression top-k heaps, counter-sample weights, per-second plan-selection cost Oct 14, 2026
@zzylol
zzylol merged commit 80fc72e into main Oct 14, 2026
3 of 4 checks passed
@zzylol zzylol added CSE enhancement New feature or request labels Oct 16, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

CSE enhancement New feature or request

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant