[zzylol]

Why

A runtime that measures its own costs could only state sketch update, merge and estimate times, and only for exact parameter matches. Exact accumulators, scans (raw reads), aggregates, per-request overhead, phase differences and memory had no slot. Measuring the KLL but not the exact plan's operators made the exact rebuild look cheap, so Stage 3 flipped quantile queries to it (#771). InstallablePlan also had no prediction to compare with what the runtime measured.

What

• PlanSelectionSettings.measured_operation_costs is now a MeasuredOperationCosts table. Each OperatorCost entry is keyed by operator kind (every physical operator), phase (ingestion_time, query_time, or both), and family (sketch params, exact kind, or aggregate measure). Its value is ns_per_unit + ns_per_invocation, with optional memory_bytes_per_state. The table also has ns_per_evaluation per query request and provenance (executor version, machine).
• When a sketch's parameters match no entry exactly, it is priced between the two nearest measured configurations of the same sketch. The interpolation is linear in log state size. Outside the measured range it takes the nearest entry.
• A node with no entry gets its formula CPU operations × one calibration factor; its scanned bytes are not scaled, so its CPU never goes negative. The factor is Σ measured ÷ Σ formula time, over the measured nodes of all candidates. It is one global heuristic, not a measurement, so a partial table can still mis-rank plans (see After). Only a table that covers every operator kind the candidates use makes the choice the measured one. A per-operator-kind factor is the follow-up, A partial measured cost table can still mis-rank plans: calibrate per operator kind #791.
• A query-time Scan entry prices raw samples fetched and decoded from the external database. That cost counts in the candidate's total.
• SelectedPlanReport.predicted_usage carries Stage 3's predicted usage for the selected plan: the totals, plus per-node and per-query figures. Each node names the installed DAG's nodes that run it: a pane chain lists its expanded panes, every other node its own id.
• The earlier JSON array of MeasuredOperationCost still loads. Each entry becomes its operator's entry at both phases, and an update becomes both a SummaryAgg and a PaneChain entry.

How

price is now price_at_scale(..., formula_scale). It computes each node's formula estimate and then looks the node up in the table:

• With an entry: the node gets units × ns_per_unit + ns_per_invocation, scanning included. Units are samples for a scan, states for a merge, and input rows otherwise.
• Without an entry: the node gets its formula CPU operations × formula_scale; its scanned bytes stay as the formula gives them.

price_at_scale also returns the measured and formula time of its measured nodes. select_plan runs the pre-pricing checks, prices every checked candidate once at scale 1 to fit the factor, then prices again at that factor for the latency and capacity checks and the ranking. Rejections stay in candidate order. A table with no operators skips the first pass, so results with an empty table are unchanged.

The compaction merge uses the state's query-time SummaryMerge entry, else one formula operation × the factor. Per-request cost is added to the CPU of the node the query reads first, and to the query's latency.

Key code interfaces added / modified

crates/plan-selection/src/measured_operation_costs.rs

/// The deployment's measured cost table (stays; was Vec<MeasuredOperationCost>).
/// JSON: this object, or the earlier array of MeasuredOperationCost.
pub struct MeasuredOperationCosts {                // new
    /// One entry per (operator, phase, family) measured; default empty.
    pub operators: Vec<OperatorCost>,              // new
    /// CPU ns per query request (scheduling, request handling, encoding); default 0.
    pub ns_per_evaluation: f64,                    // new
    /// Executor version and machine measured on; recorded, not read.
    pub provenance: Option<CostTableProvenance>,   // new
}
/// Where the numbers come from.
pub struct CostTableProvenance {                   // new
    /// The executor build measured.
    pub executor_version: String,
    /// The machine it ran on.
    pub machine: String,
}
/// One measured operator.
pub struct OperatorCost {                          // new
    /// The operator kind.
    pub operator: OperatorKind,
    /// ingestion_time / query_time; None = both (a phase entry wins).
    pub phase: Option<ExecutionTiming>,
    /// Sketch family; other params of the same sketch are interpolated.
    pub sketch: Option<SketchParams>,
    /// Exact accumulator family, e.g. "Sum".
    pub exact: Option<ExactKind>,
    /// Aggregate measure kind, e.g. "quantile".
    pub measure: Option<String>,
    /// CPU ns per unit (sample, row, state, group).
    pub ns_per_unit: f64,
    /// CPU ns per computation; default 0.
    pub ns_per_invocation: f64,
    /// Kernel memory_bytes() per state; None = formula bytes.
    pub memory_bytes_per_state: Option<f64>,
}
/// Every physical operator, by its DAG name (Scan, Filter, Project, Aggregate, Join, SetOp,
/// Concat, Dedup, Sort, Limit, BinaryOp, SQLWindowFunc, TimeRange, TimeShift,
/// PromqlVectorFromScalar, PromqlRelabel, PromqlInfoEnrich, PromqlSeriesSample, PromqlSubquery,
/// Values, SummaryAgg, SummaryEstimate, FinalizeExactAccumulator, MaintainPopulation,
/// EvaluatePopulation, SummaryMerge, PaneChain, SummarySubtract, SummaryDelete, SummaryJoin, Extension).
pub enum OperatorKind { /* the 31 variants listed above */ }   // new
impl OperatorKind { pub fn of(payload: &PhysicalASAPOperatorPayload) -> Self; }  // new
/// Earlier-format entry (stays, loadable only through MeasuredOperationCosts).
pub struct MeasuredOperationCost { pub sketch: SketchParams, pub operation: SummaryOperation, pub cpu_ns_per_operation: f64 } // unchanged
impl From<Vec<MeasuredOperationCost>> for MeasuredOperationCosts;  // new
/// Now also Deserialize.
pub enum NodePriceSource { Analytical, Measured }  // unchanged variants

crates/plan-selection/src/lib.rs

pub struct PlanSelectionSettings {
    pub measured_operation_costs: MeasuredOperationCosts, // was Vec<MeasuredOperationCost>
    /* cost_model, resource_speed, non_repeating_query_horizon_seconds: unchanged */
}
pub struct NodeCost {
    /// Its operator kind.
    pub operator: OperatorKind,                    // new
    /// Ingestion or query time.
    pub timing: ExecutionTiming,                   // new
    /// Core-seconds of CPU per second of wall time.
    pub cpu_core_seconds_per_second: f64,          // new
    pub cost_per_second: Option<f64>,              // unchanged
    pub multiplier: u64,                           // unchanged
    pub detail: String,                            // unchanged
    pub price_source: NodePriceSource,             // unchanged
}

crates/plan-selection/src/cost/analytical_cost.rs

pub enum AnalyticalCostError {
    /// Bad input: two entries for one operator, phase and family.
    DuplicateOperatorCost(String),                 // was DuplicateMeasuredOperationCost { sketch, operation }
    /// Bad input: an entry names more than one family.
    AmbiguousOperatorCost(String),                 // new
    /* other variants unchanged */
}

crates/planner/src/installable.rs

pub struct SelectedPlanReport {
    pub selected_candidate_id: String,             // unchanged
    pub cost_per_second: Option<f64>,              // unchanged
    /// Stage 3's predicted usage of the selected plan; None without a recorded cost.
    pub predicted_usage: Option<PredictedUsage>,   // new
}
/// Predicted usage per second of wall time.
pub struct PredictedUsage {                        // new
    /// Σ per_node CPU core-seconds per second.
    pub cpu_core_seconds_per_second: f64,
    /// Bytes scanned per second.
    pub scanned_bytes_per_second: f64,
    /// Average memory held.
    pub average_memory_bytes: f64,
    /// Peak memory.
    pub peak_memory_bytes: u64,
    /// Disk bytes, raw data included.
    pub disk_bytes: u64,
    /// Cores for the peak.
    pub peak_cpu_cores: u32,
    /// One per node of Stage 3's (unexpanded) DAG.
    pub per_node: Vec<PredictedNodeUsage>,
    /// One per query, workload order.
    pub per_query: Vec<PredictedQueryUsage>,
}
/// One node's prediction.
pub struct PredictedNodeUsage {                    // new
    /// Id in Stage 3's DAG (a pane chain is one node there).
    pub stage3_node_id: PhysicalASAPNodeId,
    /// Ids in the installed `dag` that run it: a pane chain's panes, newest
    /// first; any other node, its own id; empty for a node the expansion
    /// rebuilds once per pane.
    pub installed_node_ids: Vec<PhysicalASAPNodeId>,
    /// Operator kind.
    pub operator: OperatorKind,
    /// Phase.
    pub timing: ExecutionTiming,
    /// CPU core-seconds per second.
    pub cpu_core_seconds_per_second: f64,
    /// Price per second under cost model 1.
    pub cost_per_second: Option<f64>,
    /// Measured or formula.
    pub price_source: NodePriceSource,
}
/// One query's latency per evaluation.
pub struct PredictedQueryUsage {                   // new
    /// Newest-window compaction ms.
    pub compaction_ms: f64,
    /// Query-time work ms, per-request cost included.
    pub query_ms: f64,
}

crates/types/src/ir/operator/pane_chain.rs

pub struct PaneChainExpansion {
    pub roots: Vec<QueryRoot>,                    // unchanged
    pub chains: Vec<ExpandedPaneChain>,           // unchanged
    /// Every other node reached, with the node replacing it (itself when unchanged).
    pub expanded_nodes: Vec<(Rc<OperatorNode>, Rc<OperatorNode>)>, // new
}

asap_planner also re-exports MeasuredOperationCosts, OperatorCost, OperatorKind, CostTableProvenance, NodePriceSource, ExecutionTiming and the Predicted* types.

Usage

// What the deployment's runtime measured, in its settings file.
let settings: PlanSelectionSettings = serde_json::from_str(r#"{
  "measured_operation_costs": {
    "operators": [
      {"operator": "Scan", "phase": "ingestion_time", "ns_per_unit": 20.0},
      {"operator": "Scan", "phase": "query_time", "ns_per_unit": 200.0},
      {"operator": "PaneChain", "sketch": {"Kll": {"k": 200}}, "ns_per_unit": 120.0},
      {"operator": "PaneChain", "sketch": {"Kll": {"k": 1024}}, "ns_per_unit": 200.0},
      {"operator": "Aggregate", "measure": "quantile", "ns_per_unit": 10.0}
    ],
    "ns_per_evaluation": 20000.0,
    "provenance": {"executor_version": "0.1.0", "machine": "x86_64 16 cores"}}}"#)?;
// Stage 3 reads it through plan_installable: a KLL sized k = 469 interpolates
// between 120 and 200 ns; unmeasured nodes take formula x calibration factor.
let plan = plan_installable(UserInput::new(&workload, FrontendInput::Promql { histograms: None },
    PlanningModels::builtin().with_plan_selection(&settings))).await?;
// The runtime reports this next to what it measures.
let predicted = plan.selection.unwrap().predicted_usage.unwrap();
println!("{} core-s/s", predicted.cpu_core_seconds_per_second);

The end-to-end flows are checked in calibration_keeps_unmeasured_nodes_on_the_measured_scale, external_raw_reads_make_the_query_time_raw_plan_costlier (plan-selection) and the_plan_carries_stage3s_predicted_usage (planner runtime_install).

Before this PR

The test fixture is a per-series p99 over 5 min, queried every minute, 10,000 rows/s, priced on CPU only. The candidates are EXACT (rebuild from raw samples at query time), B2 (P1-m0, KLL panes rebuilt at query time) and B1 (P1-m1, KLL panes maintained at ingestion time). The table measures the KLL as expensive (150 ns per row, 2,000 ns per merge, 5,000 ns per estimate) and the time range as cheap, but not the scan or the aggregate.

Without calibration, which is how the table used to behave, the predicted CPU in core-seconds over 90 minutes is:

Candidate	Predicted (partial table, no calibration)	Under the full table (scan 100 ns, quantile 50 ns)
EXACT	1.78 (selected)	41.04
B2	54.70	81.00
B1	21.74	27.00

The exact rebuild looks 23x cheaper than it really is, so it is selected. A KLL sized k = 469 could not use a k = 200 or k = 1024 entry. A query-time raw read had no price. The plan file carried no prediction.

After this PR

With the same partial table, the calibration factor scales the unmeasured scan and aggregate:

Candidate	Predicted (partial table, calibrated)	Under the full table
EXACT	102.16	41.04
B2	111.44	81.00
B1	33.09 (selected)	27.00 (selected)

B1 is selected, which is also the choice under the full table.

The calibration factor can still mis-rank plans. Suppose the unmeasured scan and quantile really cost 40 and 20 ns. The full table then selects EXACT, but the partial table still prices them at the KLL's ratio and selects B1, which costs 1.42× more than EXACT under the full table. The gap grows as those costs fall; review measured up to 2.6×. a_partial_table_can_still_mis_rank_plans pins this down, and #791 tracks the per-operator-kind factor. With a full table where the exact plan really is cheaper (1 ns per sample and per row), EXACT is selected. Pricing query-time raw reads at 200 ns per sample turns the choice from EXACT (when raw reads are free) to B1. The plan file carries predicted_usage. For Example 4b, a pane-only plan, its per-node prices add up to Stage 3's selection cost.

These numbers come from the plan-selection fixture. The ASAPQuery-backend harness numbers in #771 need its measure_costs extended to every operator kind, which is the runtime-side follow-up.

Verification

• cargo test --workspace --locked passes.
• New tests:
  • calibration_keeps_unmeasured_nodes_on_the_measured_scale
  • a_partial_table_can_still_mis_rank_plans
  • a_calibration_factor_below_one_never_makes_cpu_negative (regression test for negative CPU; failed before the fix with cpu_ops_per_second: -466180.0)
  • a_kll_between_two_measured_sizes_is_interpolated
  • external_raw_reads_make_the_query_time_raw_plan_costlier
  • the_plan_carries_stage3s_predicted_usage
  • a_measured_entry_prices_its_node_and_calibrates_the_others
  • a_phase_entry_wins_over_one_for_both_phases
  • each_request_pays_the_per_evaluation_cost
  • measured_state_bytes_size_the_kept_state
  • the_earlier_table_format_still_loads
  • invalid_measured_tables_are_rejected
  • compaction_prices_merges_from_the_measured_merge (updated)
• Not run on this branch: clippy, and the CI wait.

Limitations: the calibration factor is a single global heuristic (#791). A node that the expansion rebuilds once per pane (the chain's input window) has no installed id. convert_rqe_atomic_costs still writes the earlier format, which loads.

Closes #771