feat(planner)!: Stage 3 resource usage and cost models 1-3 (usage over time, provisioned capacity, own cluster)

[zzylol]

Advances #683 (cost models 1 and 2, and Example 4's different choices) and #660 (cost model 3).

Why

Plan selection (Stage 3) had one cost formula: price CPU operations, scanned bytes and kept-state byte-seconds per second, and take the cheapest valid candidate. That fits pay-as-you-go billing (cost model 1). It does not fit a deployment that pays for provisioned capacity (cost model 2, #683), or one that runs on its own cluster with fixed memory and disk and wants the lowest latency or a trade-off (cost model 3, #660). Prices, speeds and the one-off horizon also lived in one struct (ResourceCalibration), so a model that has no prices could not be expressed.

What

• Stage 3 measures each valid candidate's resource usage with no prices: CPU operations and scanned bytes per second, average and peak memory, disk, peak CPU cores and each query's latency (ResourceUsage). Each node has the fields that add up over nodes (NodeUsage).
• The deployment chooses a cost model (CostModel) in PlanSelectionSettings, which PlanningModels carries as plan_selection:
  • UsageOverTime (cost model 1, the default): unit cost × usage per second; the cheapest wins. With the illustrative unit costs it reproduces every previous price bit for bit.
  • ProvisionedCapacity (cost model 2): unit cost × peak memory, peak CPU cores and disk; the cheapest wins.
  • ClusterCapacity (cost model 3): no prices. Memory, disk and (optionally) CPU cores are hard limits, so a candidate over one is invalid with the reason. The objective LowestLatency, LeastMemory or ParetoFront chooses among the rest; Selection.front lists the Pareto front.
• On docs: propose workload-wide planning, summary sharing, and materialization #509 Example 4 (Pattern B), the three models choose three different plans (numbers below).
• stage_pipeline --cost-model ... exposes the choice. Its document now reports each shown candidate's usage and the front. The candidate's total and each node's cost are renamed cost_per_second, as in Rust. A candidate's is null under cost model 3; a node's is null under cost models 2 and 3, which price or limit the whole candidate's peaks. The viewer accepts a null.
• Only devtools (run_planning_stages_over, stage_pipeline) sets plan_selection today. The planner crate and Stage1ToStage3Pass run with the default, cost model 1.

How

• price walks the physical DAG as before, but for each node it records usage instead of a price:
  • CPU operations and scanned bytes are one computation's figures times its rate.
  • Average memory is the kept state's byte-seconds per second. Ingestion-time state counts every second; query-time state counts only while its evaluation runs. This is exactly what cost model 1 charged before.
  • Peak memory assumes every evaluation overlaps: in-memory kept state, plus each computation's working memory once per consumer that recomputes it. A state maintained in memory by k ingest workers adds the k − 1 workers' partial copies of its newest window, which are merged at query time. These copies count only toward peak memory; average memory, and so cost model 1, is unchanged.
  • Disk is state on Medium::Disk, plus the raw data query-time scans read: per source, the longest window scanned. This changes what Medium::Disk means. On main, MatAtQuery(Disk) and MatAtIngestion(Disk) were priced as memory (retained byte-seconds); now their bytes are disk at the disk price and add no memory. Stage 2 places nothing on disk yet, so no output changes.
  • Peak cores are the ingestion work's cores plus one core per query evaluating at once (at least the cores its average query work fills).
• Cost model 1 prices each node as it is measured, in the same arithmetic order as before, so totals are bit-identical. Cost model 2 prices the candidate's peaks. Cost model 3 leaves cost as None.
• select_plan validates the settings once, before any candidate (SelectionError::InvalidSettings). It checks only what the chosen model reads: the speed, the horizon, and cost model 1's usage prices or cost model 2's provisioned and disk prices. A price the model does not read is not checked. If every price the model reads is 0, the settings are rejected, so cost model 1 with only provisioned prices set does not price everything at 0.
• Selection keeps "first wins on ties" for every model. Cost models 1 and 2 compare price.
  • LowestLatency compares worst-query latency, then peak memory, then disk.
  • LeastMemory compares peak memory, then disk, then latency.
  • ParetoFront keeps the candidates that no other candidate beats on all three, and selects the front's lowest-latency member, choosing only among front members.
  • Price and latency tie within rounding (1e-9 relative). Peak memory and disk are whole bytes, compared exactly in both ranking and dominance, so a candidate dominated by a few bytes is never selected off the front.
• Where horizon_s went: it is the horizon in seconds over which Stage 3 spreads non-repeating queries (one-off, scheduled and unknown). It is now PlanSelectionSettings::non_repeating_query_horizon_seconds, next to the cost model and speed, because it is a deployment choice for plan selection and not a price.
• Name clash: the enum is stage3_plan_selection::CostModel. The legacy pub trait CostModel stays at cost_model::CostModel, re-exported at the crate root. No module imports both, so neither one needed a path or a rename.
• ResourceCalibration stays for the legacy core (PhysicalPlanCostModel) only. The fields only Stage 3 read are deleted, as is DEFAULT_HORIZON_S.

Key code interfaces added / modified

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

/// Cost per second of wall time; the unit of cost models 1 and 2.
pub const COST_UNIT: &str = "cost_per_second";                       // unchanged
/// Names the analytical model behind every cost.
pub const COST_SOURCE: &str = "analytical-cost-v2 (illustrative statistics)"; // unchanged
/// Seconds non-repeating (one-off, scheduled, unknown) queries are spread over by default.
pub const DEFAULT_NON_REPEATING_QUERY_HORIZON_SECONDS: f64 = 3_600.0; // new; was analytical_cost::DEFAULT_HORIZON_S
/// The prices plan selection used until a deployment is calibrated.
pub fn illustrative_calibration() -> ResourceCalibration;            // removed: ResourceUnitCost::illustrative + ResourceSpeed::illustrative

/// What one candidate uses, per second of wall time unless the name says otherwise,
/// without prices: its nodes' NodeUsage summed, with its peak cores and latency.
/// Produced by Stage 3 for every valid candidate; every cost model reads it. New type.
pub struct ResourceUsage {
    /// CPU operations per second: ingestion-time and query-time work.
    pub cpu_ops_per_second: f64,                 // new
    /// Bytes scanned per second.
    pub scanned_bytes_per_second: f64,           // new
    /// Memory held between computations, averaged over time (byte-seconds per second).
    pub average_memory_bytes: f64,               // new
    /// The most memory live at once, assuming every evaluation overlaps.
    pub peak_memory_bytes: u64,                  // new
    /// Bytes on disk: state on Medium::Disk plus raw data query-time scans read.
    pub disk_bytes: u64,                         // new
    /// CPU cores to provision for the peak: ingestion cores plus one per query evaluating.
    pub peak_cpu_cores: u32,                     // new (not in the approved design; cost model 2 prices it)
    /// One per query, in workload order: its latency per evaluation.
    pub latency_per_query: Vec<QueryLatency>,    // new; was CandidateCost::latency
}

/// What one node uses: the ResourceUsage fields that add up over nodes. New type.
pub struct NodeUsage {
    /// CPU operations per second.
    pub cpu_ops_per_second: f64,       // new
    /// Bytes scanned per second.
    pub scanned_bytes_per_second: f64, // new
    /// Kept memory averaged over time (byte-seconds per second).
    pub average_memory_bytes: f64,     // new
    /// The most memory the node holds at once.
    pub peak_memory_bytes: u64,        // new
    /// Bytes the node keeps on disk.
    pub disk_bytes: u64,               // new
}

/// The price of one unit of each resource, per second of wall time; a deployment input.
/// New type; replaces Stage 3's use of ResourceCalibration.
pub struct ResourceUnitCost {
    /// Price of one CPU operation (cost model 1).
    pub cost_per_cpu_op: f64,                                  // new; was ResourceCalibration::cost_per_cpu_op
    /// Price of scanning one byte (cost model 1).
    pub cost_per_scanned_byte: f64,                            // new; was ResourceCalibration::cost_per_scan_byte
    /// Price of holding one byte of memory for one second (cost model 1).
    pub cost_per_memory_byte_second: f64,                      // new; was ResourceCalibration::cost_per_retained_byte_second
    /// Price of provisioning one byte of memory for one second (cost model 2).
    pub cost_per_provisioned_memory_byte_second: f64,          // new
    /// Price of provisioning one CPU core for one second (cost model 2).
    pub cost_per_provisioned_cpu_core_second: f64,             // new
    /// Price of keeping one byte on disk for one second (cost models 1 and 2).
    pub cost_per_disk_byte_second: f64,                        // new
    // removed: `version: String` (in the first head; nothing outside tests read it)
}
impl ResourceUnitCost {
    /// 1e-6 per op, 1e-7 per scanned byte, 1.25e-7 per memory byte-second (held or
    /// provisioned), 1,000 per provisioned core-second, disk unpriced (0).
    pub fn illustrative() -> Self;                                           // new
}

/// How fast one core does work; the latency check and peak cores read it. New type.
pub struct ResourceSpeed {
    /// Milliseconds one CPU operation takes on one core.
    pub ms_per_cpu_op: f64,          // new; was ResourceCalibration::ms_per_cpu_op
    /// Milliseconds scanning one byte takes on one core.
    pub ms_per_scanned_byte: f64,    // new; was ResourceCalibration::ms_per_scan_byte
}
impl ResourceSpeed {
    /// 1e-6 ms per op and 1e-7 ms per scanned byte (one cost unit is one core-millisecond).
    pub fn illustrative() -> Self;   // new
}

/// How plan selection turns valid candidates' usage into a choice; chosen by the deployment.
/// New type.
pub enum CostModel {
    /// Cost model 1: Σ unit cost × usage per second; cheapest valid wins.
    UsageOverTime(ResourceUnitCost),        // new
    /// Cost model 2: unit cost × peak memory, peak CPU cores and disk; cheapest valid wins.
    ProvisionedCapacity(ResourceUnitCost),  // new
    /// Cost model 3: no prices; the limits are hard, the objective chooses.
    ClusterCapacity(ClusterLimits),         // new
}

/// A fixed cluster's capacity (cost model 3) and how to choose within it. New type.
pub struct ClusterLimits {
    /// Memory available; a candidate whose peak memory is above it is invalid.
    pub memory_bytes: u64,                  // new
    /// Disk available; a candidate keeping more on disk is invalid.
    pub disk_bytes: u64,                    // new
    /// CPU cores available; None leaves cores unchecked.
    pub cpu_cores: Option<u32>,             // new
    /// How to choose among the candidates that fit.
    pub objective: ClusterObjective,        // new
}

/// What cost model 3 optimizes among the candidates that fit. New type.
pub enum ClusterObjective {
    /// Smallest worst-query latency; ties by peak memory, then disk.
    LowestLatency,   // new
    /// Smallest peak memory; ties by disk, then latency.
    LeastMemory,     // new
    /// Report the front over peak memory, disk and latency; select its lowest-latency member.
    ParetoFront,     // new
}

/// The deployment's plan selection choices; read from PlanningModels::plan_selection. New type.
pub struct PlanSelectionSettings {
    /// How candidates are compared; default UsageOverTime(ResourceUnitCost::illustrative()).
    pub cost_model: CostModel,                     // new
    /// Time per op and per scanned byte; default ResourceSpeed::illustrative().
    pub resource_speed: ResourceSpeed,             // new
    /// Seconds non-repeating (one-off, scheduled, unknown) queries are spread over; default 3,600.
    pub non_repeating_query_horizon_seconds: f64,  // new; was ResourceCalibration::horizon_s
}
impl Default for PlanSelectionSettings;            // new
impl PlanSelectionSettings {
    /// Bad input: bad speed, a horizon that is not positive, or a price the chosen
    /// model reads that is negative or not finite, or all of them 0.
    pub fn validate(&self) -> Result<(), AnalyticalCostError>;   // new
}

/// One node's usage and price per second. The type stays.
pub struct NodeCost {
    /// The node's usage.
    pub usage: NodeUsage,                 // new
    /// Its price per second under cost model 1; None under cost models 2 and 3.
    pub cost_per_second: Option<f64>,     // was `cost: f64`
    /// Consumers it is computed for (1 when materialized).
    pub multiplier: u64,            // unchanged
    /// Times it is computed per second.
    pub per_second: f64,            // unchanged
    /// One line on how it was measured (e.g. "scan 20000000 samples; x 0.0167/s").
    pub detail: String,             // unchanged; adds "keeps N bytes of raw data on disk" to query-time scans
}

/// One candidate's usage and price over the whole workload. The type stays.
pub struct CandidateCost {
    /// The candidate's usage: its nodes' sums, its peak cores and per-query latency.
    pub usage: ResourceUsage,                              // new
    /// Price per second under cost model 1 or 2; None under cost model 3.
    pub cost_per_second: Option<f64>,                      // was `total: f64`
    /// COST_UNIT.
    pub unit: &'static str,                                // unchanged
    /// COST_SOURCE.
    pub source: &'static str,                              // unchanged
    /// One entry per DAG node.
    pub per_node: BTreeMap<PhysicalASAPNodeId, NodeCost>,  // unchanged
    // removed: `latency: Vec<QueryLatency>` (now usage.latency_per_query)
}

/// One query's latency per evaluation. The type stays.
pub struct QueryLatency {
    /// Milliseconds to compact the newest window of an ingestion-time state the query reads.
    pub compaction_ms: f64,   // unchanged
    /// Milliseconds of the query-time work the query reaches.
    pub query_ms: f64,        // unchanged
}

/// Stage 3's result. The type stays.
pub struct Selection {
    /// The selected candidate's id.
    pub selected: String,                                              // unchanged
    /// The selected candidate's roots with confirmed weight requirements.
    pub plan: Vec<QueryRoot>,                                          // unchanged
    /// Usage and price of every valid candidate.
    pub costs: BTreeMap<String, CandidateCost>,                        // unchanged
    /// Per valid candidate and query, the accuracy accepted.
    pub accepted: BTreeMap<String, Vec<AcceptedAccuracy>>,             // unchanged
    /// Every candidate not selected: invalid (with a reason) or valid but costlier,
    /// slower, larger or off the front.
    pub rejected: Vec<Rejection>,                                      // unchanged
    /// Under ClusterObjective::ParetoFront, the front in candidate order; else empty.
    pub front: Vec<String>,                                            // new
}

/// Why Stage 3 failed. The type stays.
pub enum SelectionError {
    /// Bad input: the plan selection settings, checked once before any candidate.
    InvalidSettings(AnalyticalCostError),                                    // new
    /// Bad input: a candidate has a different number of roots than the workload entries.
    EntryCount { candidate: String, roots: usize, entries: usize },          // unchanged
    /// A node could not be measured (for example, a cost overflowed).
    Cost { candidate: String, node: PhysicalASAPNodeId, error: AnalyticalCostError }, // unchanged
    /// Every candidate was rejected; each with its reason.
    NoValidCandidate(Vec<Rejection>),                                        // unchanged
}

/// Stage 3: reject invalid candidates (accuracy, latency bound and, under cost model 3, the
/// cluster's limits), measure the rest and choose under models.plan_selection.cost_model.
pub fn select_plan(
    cands: &[PhysicalCandidate],
    entries: &[QueryWorkloadEntry],
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Result<Selection, SelectionError>;   // unchanged signature; reads models.plan_selection

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

/// Planning logic the stages read. The type stays.
pub struct PlanningModels<'a> {
    /// The legacy cost model trait (not read by the planning stages).
    pub cost: &'a dyn CostModel,                     // unchanged (crate::cost_model::CostModel, the trait)
    /// The built-in or deployment accuracy model.
    pub accuracy: &'a dyn AccuracyModel,             // unchanged
    /// The deployment's empirical accuracy evidence.
    pub evidence: &'a dyn AccuracyEvidenceProvider,  // unchanged
    /// The deployment's plan selection settings; default PlanSelectionSettings::default().
    pub plan_selection: &'a PlanSelectionSettings,   // new
}
impl<'a> PlanningModels<'a> {
    /// These models with the default plan selection settings.
    pub fn new(cost: &'a dyn CostModel, accuracy: &'a dyn AccuracyModel,
               evidence: &'a dyn AccuracyEvidenceProvider) -> Self;              // unchanged
    /// The built-in models, with cost model 1 and the illustrative unit costs.
    pub fn builtin() -> PlanningModels<'static>;                                 // unchanged
    /// These models with another legacy cost model.
    pub fn with_cost(self, cost: &'a dyn CostModel) -> Self;                     // unchanged
    /// These models with another accuracy model.
    pub fn with_accuracy(self, accuracy: &'a dyn AccuracyModel) -> Self;         // unchanged
    /// These models with another accuracy evidence provider.
    pub fn with_evidence(self, evidence: &'a dyn AccuracyEvidenceProvider) -> Self; // unchanged
    /// These models with the deployment's plan selection settings.
    pub fn with_plan_selection(self, plan_selection: &'a PlanSelectionSettings) -> Self; // new
}

Only devtools calls with_plan_selection today (run_planning_stages_over, so stage_pipeline). The planner crate's passes use PlanningModels::builtin(), which is cost model 1.

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

/// The legacy core's prices (PhysicalPlanCostModel); Stage 3 uses ResourceUnitCost.
/// The type stays.
pub struct ResourceCalibration {
    /// Price of one CPU operation.
    pub cost_per_cpu_op: f64,                  // unchanged
    /// Price of scanning one byte.
    pub cost_per_scan_byte: f64,               // unchanged
    /// Price of one byte of peak memory per computation.
    pub cost_per_retained_byte: f64,           // unchanged
    /// Names the price set.
    pub version: String,                       // unchanged
    // removed (read only by Stage 3, now elsewhere): cost_per_retained_byte_second,
    // horizon_s, ms_per_cpu_op, ms_per_scan_byte
}
pub const DEFAULT_HORIZON_S: f64;              // removed: stage3_plan_selection::DEFAULT_NON_REPEATING_QUERY_HORIZON_SECONDS

crates/devtools/src/planning_stages.rs

/// Stages 1-3 over roots lowered from any query language; plan selection compares
/// candidates under `plan_selection`'s cost model.
pub fn run_planning_stages_over(
    roots: Vec<QueryRoot>,
    entries: &[QueryWorkloadEntry],
    data: &DataWorkload,
    max_combinations: Option<usize>,
    plan_selection: &PlanSelectionSettings,   // new parameter
) -> Result<PlanningStagesRun, PlanningStagesError>;

crates/devtools/src/bin/stage_pipeline.rs: new flags --cost-model usage-over-time|provisioned-capacity|cluster-capacity, --cluster-memory-bytes, --cluster-disk-bytes, --cluster-cpu-cores, --cluster-objective lowest-latency|least-memory|pareto-front. The document gains stage3_plan_selection.costs.<id>.usage (cpu_ops_per_second, scanned_bytes_per_second, average_memory_bytes, peak_memory_bytes, disk_bytes, peak_cpu_cores, latency_per_query as [{compaction_ms, query_ms}]) and stage3_plan_selection.front. A candidate's total and a node's cost are renamed cost_per_second and are null under cost model 3.

Usage, end to end

#509 Example 4, Pattern B, planned under each cost model. This was compiled and run as a test at the head commit, and every assertion passed.

// 1. What the user writes: #509 Example 4, Pattern B. One p99 per series over the last
//    5 minutes, `quantile_over_time(0.99, latency_ms[5m])`, every minute, within rank
//    error 0.01 at δ = 0.01 and 200 ms; 1M series arriving at 66,667 samples/s.
let workload = pattern_b();
// `entries`: the query's workload entry (target, latency bound, recurrence), read by Stages 2-3.
let entries: Vec<_> = workload.query_workload.entries().collect();
// `data`: the declared data facts (series, ingestion rate) Stage 3 measures usage from.
let data = workload.data_workload.clone().unwrap();
// `roots`: Stage 0's (query-language frontends) plan root for the query, which Stage 1 reads.
let roots = lower_promql_roots(&workload).unwrap();

// 2. What the deployment declares: how plan selection compares valid candidates.
//    `pay_as_you_go`: cost model 1 with the illustrative prices, the default.
let pay_as_you_go = PlanSelectionSettings::default();
//    `provisioned`: cost model 2, pricing peak memory, peak cores and disk per second.
let provisioned = PlanSelectionSettings {
    cost_model: CostModel::ProvisionedCapacity(ResourceUnitCost::illustrative()),
    ..PlanSelectionSettings::default()
};
//    `own_cluster`: cost model 3, a 256 GB / 1 TB cluster wanting the lowest latency.
let own_cluster = PlanSelectionSettings {
    cost_model: CostModel::ClusterCapacity(ClusterLimits {
        memory_bytes: 256_000_000_000,
        disk_bytes: 1_000_000_000_000,
        cpu_cores: None,
        objective: ClusterObjective::LowestLatency,
    }),
    ..PlanSelectionSettings::default()
};

// 3. Stages 1-3 under each. Stages 1-2 give the same four valid candidates every time
//    (P1 exact, P2 KLL over the whole window, P3 = B2 five 1-min KLL panes rebuilt per
//    evaluation, P3-m1 = B1 the panes maintained at ingestion time); Stage 3's `price`
//    measures each one's `ResourceUsage` once, and the cost model chooses.
// `select`: the selection (selected id, `costs` with usage per candidate, rejections).
let select = |settings: &PlanSelectionSettings| {
    run_planning_stages_over(roots.clone(), &entries, &data, None, settings)
        .unwrap()
        .stage3_plan_selection
        .unwrap()
};
// `cost_model_1`: cost model 1's choice; P2 costs 2.100/s, B1 13,326.530/s (107 GB of
// panes held every second).
let cost_model_1 = select(&pay_as_you_go);
assert_eq!(cost_model_1.selected, "P2");
assert_eq!(cost_model_1.costs["P3-m1"].usage.average_memory_bytes, 106_608_000_000.0);
// `cost_model_2`: cost model 2's choice; one core each, so peak memory decides: P1 192 MB
// (1,024.000/s) against P2's 1M KLL states, 17.8 GB at once (3,221.000/s).
let cost_model_2 = select(&provisioned);
assert_eq!(cost_model_2.selected, "P1");
assert_eq!(cost_model_2.costs["P1"].usage.peak_memory_bytes, 192_000_064);
assert_eq!(cost_model_2.costs["P2"].usage.peak_memory_bytes, 17_768_000_096);
// `cost_model_3`: cost model 3's choice; no prices (`cost` is None), B1 fits in 124 GB and
// answers in 7 ms against P2's 126 ms.
let cost_model_3 = select(&own_cluster);
assert_eq!(cost_model_3.selected, "P3-m1");
assert_eq!(cost_model_3.costs["P3-m1"].cost_per_second, None);
assert_eq!(cost_model_3.costs["P3-m1"].usage.latency_per_query[0].total_ms(), 7.0);
assert_eq!(cost_model_3.costs["P2"].usage.latency_per_query[0].total_ms(), 126.0);

The same from the command line:

stage_pipeline --example planner-layering-3b --cost-model cluster-capacity \
  --cluster-memory-bytes 256000000000 --cluster-disk-bytes 1000000000000 \
  --cluster-objective pareto-front --out run.json
# stage3_plan_selection: selected "P3-m1", front ["P1", "P2", "P3-m1"],
# P3 "dominated by P2: latency 131.0 ms, peak memory 106608000064 bytes, ..."

Before this PR

Plan selection had a single cost formula: cost model 1 with the illustrative prices, fixed in code (illustrative_calibration()). A deployment could not say it pays for provisioned capacity, or that it has a fixed cluster. Only the scalar total and the latency were reported, not the memory, disk and cores behind them. On Example 4 every deployment got P2 (the whole-window KLL).

After this PR

Example 4 (Pattern B)	P1 exact	P2 KLL (whole window)	P3 = B2 (panes rebuilt)	P3-m1 = B1 (panes at ingestion)	Selected
Cost model 1, cost/s	2.400	2.100	2.183	13,326.530	P2
Cost model 2, cost/s	1,024.000	3,221.000	14,326.000	17,547.000	P1
Peak memory	192 MB	17.8 GB	106.6 GB	124.4 GB
Disk (raw data)	640 MB	640 MB	640 MB	0
Peak cores	1	1	1	2
Latency	144 ms	126 ms	131 ms	7 ms
Cost model 3, 256 GB, lowest latency				selected	P3-m1
Cost model 3, 64 GB, lowest latency		selected	invalid (memory)	invalid (memory)	P2
Cost model 3, least memory	selected				P1
Cost model 3, Pareto front	front	front	dominated by P2	front	P3-m1

The default (cost model 1) is unchanged. The selected totals are bit-identical: Example 1 P1 7.94021, Example 2 P1 6.0800009, Example 3a P485 39,828.80278, Example 3b/4 P2 2.1. Every candidate's cost and every node's cost were compared against stage_pipeline built at main (f4a5b98), with no display cap, for all four examples: 171 candidates and 3,196 nodes, 0 differences, and the same selections and rejections.

The memory figures above come from each KLL's full capacity, not from the samples it holds. They are overstated for this workload, and the cost model 2 and 64 GB results depend on them (see Limitations, #726).

Verification

At head 014b5b2, rebased on main f4a5b98 (with #722):

• cargo nextest run -p asap-aware-mapping -p asap-devtools -p asap-integration-tests -p asap-planner: 930 passed, 29 skipped. New tests:
  • Stage 3 unit tests:
    • node_usage_adds_up_to_the_candidates
    • provisioned_capacity_prices_peak_memory_cores_and_disk
    • cluster_capacity_rejects_a_candidate_over_its_cores
    • cluster_capacity_rejects_a_candidate_over_its_disk
    • state_on_disk_is_counted_as_disk_not_memory
    • cluster_objectives_keep_the_earlier_of_tied_candidates
    • partial_copies_of_the_newest_window_add_to_peak_memory
    • raw_data_on_disk_is_the_longest_window_per_source
    • dominance_needs_one_strictly_better_resource
    • invalid_settings_are_rejected, which covers cost model 1 with only provisioned prices.
    • memory_differing_by_a_few_bytes_is_compared_exactly. Two candidates with equal latency and disk, at 124,000,000,100 and 124,000,000,000 bytes: the front is the smaller one, and the Pareto front, least memory and lowest latency all select it. This test failed before the fix: the larger one was selected, off the front.
  • memory_held_only_while_a_query_runs_is_not_charged now varies cost_per_memory_byte_second, the price cost model 1 reads. It asserts that the memory term equals average memory while peak memory is larger, so it can fail.
  • Example 4: stage3_plan_selection_b_cost_models_choose_differently and stage3_plan_selection_b_pareto_front_over_memory_disk_and_latency.
  • stage_pipeline: cost_model_flag_chooses_how_candidates_are_compared and cluster_flags_and_cost_model_3_go_together.
• Cost model 1 is bit-identical to main (f4a5b98): 171 candidates, 3,196 nodes, 0 differences, same selections and rejections (see After).
• python3 -m unittest test_viewer (tools/dag-viewer): 35 passed with a built stage_pipeline. Without one, 4 are skipped (31 passed, 4 skipped, as in the review's run).
• cargo fmt --all -- --check and cargo clippy --workspace --all-targets --all-features --locked -- -D warnings: clean. tools/verify_metricsql_external_consumer.sh: clean at the first head.
• The Usage above was compiled and run as a test at this head, and it passed.
• Review: an independent review at 4ca39b8 confirmed bit-identity and the Example 4 table; its blocking items were addressed in e9977ec. The re-review at e9977ec was clean, with follow-up fixes (the Pareto selection, the partial-copy window, the renames) made in 014b5b2. 014b5b2 has not been reviewed independently.

Deviations from the approved design

• CandidateCost.per_node keeps NodeCost (now holding usage: NodeUsage and cost_per_second: Option<f64>) instead of a bare ResourceUsage. stage_pipeline and the viewer show each node's multiplier and detail.
• ResourceUsage.latency_per_query: Vec<QueryLatency> keeps the compaction/query split that the latency rejection reason reports, instead of latency_ms: Vec<f64>. The JSON uses the same name and split.
• ResourceUsage.peak_cpu_cores and ResourceUnitCost.cost_per_provisioned_cpu_core_second are added, because cost model 2 prices cores and cost model 3 limits them.
• The price fields end in _second (cost_per_disk_byte_second, cost_per_provisioned_memory_byte_second), because every price is per second of wall time.
• The enum stays CostModel: no clash with the legacy trait, see How.
• horizon_s moved to PlanSelectionSettings::non_repeating_query_horizon_seconds, a deployment setting for plan selection, rather than to the workload. It spreads one-off, scheduled and unknown queries, hence the name.
• ResourceUnitCost.version (in the first head) is deleted: nothing outside tests read it. The legacy ResourceCalibration.version stays.
• Nodes carry NodeUsage, without the cores and latency that only a whole candidate has.

Limitations and follow-ups

• Sketch memory is each sketch's full capacity. A KLL state is sized 8 × kll_items(k) over all 61 levels: 17,768 bytes per state in Example 4. This mirrors the pinned asap_sketchlib KLL, which allocates that capacity up front. But each Pattern B series has only 20 samples per 5-minute window and 4 per 1-minute pane. For the values held, P2's "17.8 GB" would be 160 MB and B1's "107 GB" kept would be 192 MB. So cost model 2 selecting P1, cost model 3 at 64 GB rejecting B1 and B2, and cost model 1's 13,326.53/s for B1 all depend on this overstatement. Capping a state's size by the samples it holds, together with a growing implementation (KLLDynamic, or lazy allocation upstream), is [Bug] Stage 3 sizes a sketch's memory by its full capacity, not by the samples it holds #726.

• The illustrative unit costs are not calibrated. Disk is unpriced (0), and the provisioned core price (1,000/s, one core-second) makes cost model 2 rank mostly by cores and then memory. Plan selection cost models: usage over time (cost model 1), peak capacity (cost model 2), own-cluster trade-offs (cost model 3) #683's billing-survey presets remain open.

• The viewer hides cost model 3's reasons. It lists every valid candidate that was not selected as "costlier" without a reason, so "slower", "more memory", "dominated by …" and "not its lowest-latency member" appear only in the JSON's rejected. It also shows no ranking or usage under cost model 3.

• Peak memory and peak cores assume every evaluation overlaps. Overlapping evaluations of one query (latency above its interval) are not counted. The k − 1 partial copies built by k ingest workers are counted in peak memory only. Example 4 has one worker per state, so its figures do not change.

• Raw-data disk counts each source's longest query-time window. Retention policy beyond that is not modelled.

• PlanOutput and the viewer do not show usage or the front yet (G11). Under cost model 3 the viewer shows "no price" and no ranking.

• A selection error under cost model 3 still says "no valid candidate", with each candidate's limit reason.