feat(asap-aware-mapping)!: price sketch nodes from measured operation costs in Stage 3

[zzylol]

Closes #679.

Why

Stage 3 prices a sketch's CPU from its own formulas (e.g. "depth" operations per KLL or DDSketch update). A deployment that has measured its sketches, such as sketch-bench's rqe_atomic_costs.json, had no way to give Stage 3 those measured times. The only measured path was the legacy core's PhysicalPlanCostModel / empirical_cost, and Stage 3 does not use it.

What

• Measured table in the settings: PlanSelectionSettings::measured_operation_costs holds the measured CPU ns for one update, merge or estimate per sketch configuration.
• Which nodes it prices: sketch builds (per-group SummaryAgg and PaneChain), merges and estimates with an entry. Every other node keeps its formula price.
• Reporting: each node's NodeCost::price_source says measured or analytical.
• Converter: convert_rqe_atomic_costs turns sketch-bench's table into the measured table. Exact accumulators and Hydra rows are skipped, with the reason.
• CLI: stage_pipeline --measured-operation-costs <file> passes the table to Stage 3.
• Compaction: the newest window's compaction, (k − 1)·I merges when ingestion needs k > 1 cores, uses the state's measured merge too.
• Exact matches only: an entry matches only a node with exactly its sketch parameters (the error and failure probability a sketch was sized for are ignored). Near misses are priced from the formulas, with no interpolation.
• Candidate source: CandidateCost::source says whether a candidate's nodes were all priced from the formulas (COST_SOURCE), all measured (MEASURED_COST_SOURCE) or a mix (MIXED_COST_SOURCE).
• Legacy removal: removes physical_plan_cost_model, empirical_cost, empirical_resources and ResourceCalibration. This is a deliberate exception to "delete only when replaced". What they did beyond measured CPU (storage I/O, handoffs, cache profiles, per-node deployment evidence, provenance) is tracked in Plan selection: price storage I/O, handoffs and cache, from deployment evidence with provenance #733. storage_io, physical_handoff_cost and query_physical_lowering are kept for that issue.

What the deleted legacy code did, and where it goes

removed code	capability	after this PR
physical_plan_cost_model.rs	legacy CostModel adapter pricing replacement candidates into one objective	Stage 3 price(), which already selects plans; no other caller
	per-node physical evidence from the deployment (PlannerPhysicalPlanProvider, PhysicalEvidenceSnapshot)	#733 (a); query_physical_lowering.rs kept
	storage I/O pricing (StorageIoProfile: disk reads/writes, object GETs/PUTs)	#733 (b); storage_io.rs kept
	handoff pricing (PhysicalHandoffProfile: network and materialization bytes)	#733 (b); physical_handoff_cost.rs kept
	cache profiles (result and buffer hit ratios)	#733 (c)
empirical_cost.rs	measured CPU per sketch update, merge and read	this PR: measured_operation_costs (update, merge, estimate)
	lookup that ignores the KLL failure probability	this PR: SketchParams equality; test lookup_ignores_the_kll_failure_probability moved to Stage 3
	provenance, environment and distribution descriptors, validity window, refusing stale or ambiguous records	#733 (d)
	measured offline error (OfflineError)	#733 (g)
empirical_resources.rs	measured retained, peak, serialized, disk and scan bytes per sketch	#733 (e); the converter drops mem_bytes_per_instance
	measured build-one-instance and prepare-read CPU	#733 (e)
	measured exact-accumulator costs (ExactResourceMeasurements)	#733 (f); the converter skips exact-* rows
analytical_cost.rs ResourceCalibration, ResourceEstimate::calibrated_cost	one objective from per-dimension prices	Stage 3 CostModel / ResourceUnitCost (existing)

How

• Units: a measured time of t ns becomes t · 10⁻⁶ / ms_per_cpu_op CPU operations of ResourceSpeed. Latency (operations × ms_per_cpu_op) therefore equals the measured time, and all three cost models price it like any other operation.
• Matching nodes to entries: price() records each per-group sketch build's SketchParams. Merges and estimates inherit them from their producer. A node is then matched to an entry by sketch and operation.
• Per-node formulas:
  • build: rows × measured update;
  • pane chain: rows × (1 + measured update);
  • merge: states × measured merge;
  • estimate: groups × measured estimate;
  • compaction latency: (k − 1)·I × measured merge time, instead of one operation per merge.
• Bad input: PlanSelectionSettings::validate rejects:
  • a negative or non-finite time (InvalidCalibration);
  • two entries for one sketch and operation (DuplicateMeasuredOperationCost, naming both);
  • ms_per_cpu_op 0 with a non-empty table (MeasuredCostsWithoutCpuSpeed).

Key code interfaces added / modified

crates/asap-aware-mapping/src/stage3_plan_selection/measured_operation_costs.rs (new), crates/asap-aware-mapping/src/stage3_plan_selection.rs

/// One operation on a sketch's state, as plan selection prices it. New type.
pub enum SummaryOperation {
    /// Insert one input row (a `SummaryAgg` or `PaneChain` build).
    Update, // new
    /// Merge one input state into another (a `SummaryMerge`).
    Merge, // new
    /// Read one result out of one state (a `SummaryEstimate`, per group).
    Estimate, // new
}

/// The measured CPU time of one operation on one sketch configuration;
/// one entry of the measured table. New type.
pub struct MeasuredOperationCost {
    /// The sketch and its parameters, e.g. `{"Kll": {"k": 200}}`, matched with `SketchParams` equality.
    pub sketch: SketchParams, // new
    /// Which operation was measured.
    pub operation: SummaryOperation, // new
    /// CPU nanoseconds one such operation takes on one core.
    pub cpu_ns_per_operation: f64, // new
}

/// Which prices one node's CPU operations came from. New type.
pub enum NodePriceSource {
    /// Plan selection's formulas.
    Analytical, // new
    /// The measured table's entry for its sketch and operation.
    Measured, // new
}

/// Stage 3's settings. Type stays.
pub struct PlanSelectionSettings {
    /// How usage becomes one price (cost model 1, 2 or 3).
    pub cost_model: CostModel, // unchanged
    /// Time per operation and per scanned byte, for latency and cores.
    pub resource_speed: ResourceSpeed, // unchanged
    /// Seconds one-off, scheduled and unknown queries are spread over; default 3,600.
    pub non_repeating_query_horizon_seconds: f64, // unchanged
    /// The deployment's measured CPU time per sketch operation; default empty (every node from the formulas).
    pub measured_operation_costs: Vec<MeasuredOperationCost>, // new
}

/// One candidate's usage and price. Type stays.
pub struct CandidateCost {
    /// Its usage over the whole workload.
    pub usage: ResourceUsage, // unchanged
    /// Its price per second under cost model 1 or 2; `None` under cost model 3.
    pub cost_per_second: Option<f64>, // unchanged
    /// `cost_per_second`.
    pub unit: &'static str, // unchanged
    /// Where its prices came from: `COST_SOURCE`, `MEASURED_COST_SOURCE` or `MIXED_COST_SOURCE`.
    pub source: &'static str, // was always COST_SOURCE
    /// One price per DAG node.
    pub per_node: BTreeMap<PhysicalASAPNodeId, NodeCost>, // unchanged
}

/// `CandidateCost::source` when every node is priced from the formulas.
pub const COST_SOURCE: &str = "analytical-cost-v2 (illustrative statistics)"; // unchanged
/// `CandidateCost::source` when every node is priced from the measured table.
pub const MEASURED_COST_SOURCE: &str = "measured operation costs (illustrative statistics)"; // new
/// `CandidateCost::source` when some nodes are measured and the others are not.
pub const MIXED_COST_SOURCE: &str = "measured operation costs and analytical-cost-v2 (illustrative statistics)"; // new

/// One node's price in a candidate's cost. Type stays.
pub struct NodeCost {
    /// Its price per second under cost model 1; `None` under cost models 2 and 3, which price the candidate's peaks.
    pub cost_per_second: Option<f64>, // unchanged
    /// The number of consumers it is computed for (1 when materialized).
    pub multiplier: u64, // unchanged
    /// How it was priced, e.g. "build DDSketch into 1000 states: 300000 rows x 2000.000 measured ops per update; x 0.0003/s".
    pub detail: String, // unchanged
    /// Whether its CPU operations came from the measured table or the formulas.
    pub price_source: NodePriceSource, // new
}

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

/// The legacy core's prices; Stage 3 prices with `ResourceUnitCost`. Type removed.
pub struct ResourceCalibration { /* removed */ }
/// The legacy core's single objective from a `ResourceCalibration`.
impl ResourceEstimate { pub fn calibrated_cost(self, calibration: &ResourceCalibration) -> Result<f64, AnalyticalCostError>; } // removed
/// The legacy `CostModel` adapter over replacement candidates.
pub mod physical_plan_cost_model; // removed
/// Offline evidence artifacts and their applicability lookup.
pub mod empirical_cost; // removed
/// Measured resources per physical operator.
pub mod empirical_resources; // removed

/// Analytical and settings errors. Type stays; its other variants are unchanged.
pub enum AnalyticalCostError {
    /// Bad input: the measured table has two entries for one sketch configuration and operation.
    DuplicateMeasuredOperationCost { sketch: SketchParams, operation: SummaryOperation }, // new
    /// Bad input: a non-empty measured table with `ms_per_cpu_op` not positive.
    MeasuredCostsWithoutCpuSpeed(f64), // new
}

crates/devtools/src/rqe_atomic_costs.rs (new), crates/devtools/src/bin/convert_rqe_atomic_costs.rs (new), crates/devtools/src/bin/stage_pipeline.rs

/// A sketch-bench row that gave no entries, and why. New type.
pub struct SkippedRow {
    /// The row's `sketch` name, e.g. `exact-sum`.
    pub sketch_name: String, // new
    /// Why plan selection cannot price from it.
    pub reason: String, // new
}

/// What `convert_rqe_atomic_costs` produced. New type.
pub struct ConvertedCosts {
    /// The measured table: three entries per converted row, in row order.
    pub measured_operation_costs: Vec<MeasuredOperationCost>, // new
    /// The rows skipped, in row order.
    pub skipped_rows: Vec<SkippedRow>, // new
}

/// Converts parsed `rqe_atomic_costs.json`; errors (bad input) on a malformed row or a duplicate configuration.
pub fn convert_rqe_atomic_costs(rows: &Value) -> Result<ConvertedCosts, String>; // new

CLI: convert_rqe_atomic_costs <rqe_atomic_costs.json> <out.json> (new binary); stage_pipeline --measured-operation-costs <file> (new flag); the asap-stage-pipeline/v1 document's per-node costs gain "price_source": "analytical" | "measured".

Usage

# u-bench.json: two sketch-bench rows, one DDSketch (alpha 0.01: 2 µs per insert,
# 1 µs per merge, 0.5 µs per query) and one exact sum.
convert_rqe_atomic_costs u-bench.json u-measured.json
# stderr: skipped exact-sum: an exact accumulator: the table prices sketches only
#         wrote 3 entries to u-measured.json
# u-measured.json: [{"sketch": {"DDSketch": {"alpha": 0.01}}, "operation": "update", "cpu_ns_per_operation": 2000.0}, ...merge 1000.0, estimate 500.0]

# One quantile query. --epsilon 0.01 --delta 0.001 is its accuracy target (error 0.01,
# failure probability 0.001), so Stage 1 offers sketch candidates (here DDSketch,
# alpha 0.01) beside the exact plan. --measured-operation-costs hands Stage 3 the
# table above. u1.json is the asap-stage-pipeline/v1 document with all four stages;
# its stage3_plan_selection holds each candidate's per-node costs with price_source.
stage_pipeline --promql 'quantile_over_time(0.99, latency[5m])' --epsilon 0.01 --delta 0.001 \
  --measured-operation-costs u-measured.json --out u1.json
# u0.json: the same run without --measured-operation-costs, for comparison.

stage3_plan_selection in u1.json, compared with the same run without the flag:

candidate	without the table	with the table
P1 (exact)	0.0006 cost/s	0.0006 cost/s, all nodes analytical; source COST_SOURCE
P3 (DDSketch, alpha 0.01)	0.000517 cost/s, selected	0.167 cost/s; build "300000 rows x 2000.000 measured ops per update" and estimate "1000 groups x 500.000 measured ops per estimate" are measured; source MIXED_COST_SOURCE
selected	P3	P1; P3 rejected "costlier: 0.167 vs 0.001 cost_per_second"

Before this PR

Stage 3 priced every sketch from its formulas. Measured sketch costs could only reach the legacy PhysicalPlanCostModel, which plan selection does not call.

After this PR

A deployment converts its sketch-bench measurements once and passes the table, in code or with --measured-operation-costs. Stage 3 prices those sketches from it, so selection can change, and each node says where its price came from. With no table, every price and selection is unchanged. The only output change is price_source: "analytical" on each node.

Verification

• fmt: cargo fmt --all -- --check passes.
• clippy: cargo clippy -p asap-aware-mapping -p asap-devtools --all-targets -- -D warnings passes.
• Tests (rebased on feat(asap-aware-mapping)!: deployment capabilities (supported summaries, raw-data and query-time retention) #730): cargo nextest run -p asap-aware-mapping --lib -p asap-devtools passes 558/558, and cargo nextest run -p asap-devtools (with its integration tests) passes 39/39. The example 1 fixture was regenerated; its only change is price_source: "analytical" on its 240 nodes.
• New tests:
  • Stage 3: a measured entry prices exactly its node and reports it (and the candidate's source becomes mixed), and the candidate's CPU grows linearly in the measured time;
  • Stage 3: compaction uses the measured merge. Written first and failing before the fix, it checks that a 1,200 ns KLL merge with k > 1 ingestion workers makes compaction 1,200× the one-operation compaction;
  • Stage 3: a planner-sized KLL (failure_probability: Some(0.001)) matches a converter entry with None, and another k does not;
  • Stage 3: each invalid table gives its own error variant, and the duplicate names the sketch and operation;
  • Stage 3: entries for other sketches or operations change nothing.
  • Converter: a small sample converts and round-trips; malformed rows are errors.
• End to end: the Usage run above, with this branch's binaries.

[zzylol]

/// Which prices one node's CPU operations came from. New type.
pub enum NodePriceSource {
/// Plan selection's formulas.
Analytical, // new
/// The measured table's entry for its sketch and operation.
Measured, // new
}
what does the node refer to? A node in DAG? or a machine node?

[zzylol]

A node of the physical plan DAG (PhysicalASAPDAG): one operator such as a SummaryAgg build, a SummaryMerge or a SummaryEstimate, keyed by its PhysicalASAPNodeId in CandidateCost.per_node. It is not a machine. NodePriceSource says whether that operator's CPU operations were priced from the measured table (Measured) or from the formulas (Analytical). In the next rename PR I will make the doc comment say "one operator node of the physical plan DAG" so this is clear from the code.