From 179c82cb8b99ee0e15e24e3aec9b70d2badfb200 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 7 Sep 2026 19:08:30 -0600 Subject: [PATCH 1/4] Compare complete bound workloads using strict provider cost evidence --- .../examples/workload_cost_manifest.rs | 24 + control_plane/src/main.rs | 79 ++- control_plane/src/physical/compiler.rs | 52 +- control_plane/src/physical/mod.rs | 1 + control_plane/src/physical/workload_cost.rs | 589 ++++++++++++++++++ data_plane/src/main.rs | 6 + .../asapquery_compatibility_process_e2e.rs | 39 +- data_plane/tests/backend_process_e2e.rs | 67 ++ docs/examples/workload-cost-evidence.md | 92 +++ 9 files changed, 917 insertions(+), 32 deletions(-) create mode 100644 control_plane/examples/workload_cost_manifest.rs create mode 100644 control_plane/src/physical/workload_cost.rs create mode 100644 docs/examples/workload-cost-evidence.md diff --git a/control_plane/examples/workload_cost_manifest.rs b/control_plane/examples/workload_cost_manifest.rs new file mode 100644 index 00000000..e3463ce6 --- /dev/null +++ b/control_plane/examples/workload_cost_manifest.rs @@ -0,0 +1,24 @@ +//! Emit pricing requirements; never fabricate quotes or publish a plan. +use control_plane::physical::{ + compiler::BackendLocalPlanningSnapshot, compiler::PhysicalCompiler, workload_cost, +}; + +fn main() -> Result<(), Box> { + let path = std::env::args() + .nth(1) + .ok_or("usage: workload_cost_manifest SNAPSHOT.json")?; + let snapshot: BackendLocalPlanningSnapshot = + serde_json::from_str(&std::fs::read_to_string(path)?)?; + let (request, environment) = snapshot.planning_request()?; + let manifests = workload_cost::with_exact_alternative(request)? + .into_iter() + .map(|candidate| { + let queries = candidate.queries.clone(); + PhysicalCompiler + .compile(candidate, environment.clone()) + .and_then(|plan| workload_cost::manifest(&plan, &queries)) + }) + .collect::, _>>()?; + println!("{}", serde_json::to_string_pretty(&manifests)?); + Ok(()) +} diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 73838cc7..d2f37bf8 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -546,6 +546,10 @@ async fn main() { let app = Router::new() .route("/api/v1/plan", post(handle_plan)) + .route( + "/api/v1/physical-plan/cost-manifests", + post(handle_workload_cost_manifests), + ) .route( "/api/v1/physical-plan/compile-and-publish", post(handle_compile_and_publish_physical_plan), @@ -590,6 +594,8 @@ struct PhysicalPlanQueryRequest { #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct CompileAndPublishPhysicalPlanRequest { + #[serde(default)] + workload_cost_evidence: Option, queries: Vec, collector_ids: Vec, capability_snapshot_id: String, @@ -613,6 +619,7 @@ fn default_physical_plan_timeout_ms() -> u64 { #[derive(Debug, Serialize)] struct CompileAndPublishPhysicalPlanResponse { + cost_comparison: Option, plan_id: u64, plan_version: u64, status: &'static str, @@ -629,7 +636,7 @@ async fn handle_compile_and_publish_physical_plan( State(st): State, Json(request): Json, ) -> impl IntoResponse { - let (bundle, collector_ids, apply_timeout, adaptation_evidence) = + let (bundle, collector_ids, apply_timeout, adaptation_evidence, _) = match compile_physical_plan_request(request) { Ok(compiled) => compiled, Err(response) => return response.into_response(), @@ -711,6 +718,7 @@ async fn handle_compile_and_publish_physical_plan( } Json(CompileAndPublishPhysicalPlanResponse { + cost_comparison: bundle.cost_comparison, plan_id: bundle.envelope.plan_id, plan_version: bundle.envelope.plan_version, status: "active", @@ -731,6 +739,7 @@ fn compile_physical_plan_request( Vec, Duration, Vec, + Vec, ), (StatusCode, String), > { @@ -806,24 +815,38 @@ fn compile_physical_plan_request( return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())); } - let bundle = match physical::compiler::PhysicalCompiler.compile( - physical::compiler::PlanningRequest { - queries, - evidence: request.evidence, - planner_revision: request.planner_revision, - }, - physical::compiler::DeploymentEnvironment { - target: physical::compiler::PhysicalDeploymentTarget::DistributedCollectors, - collector_ids: request.collector_ids.clone(), - capability_snapshot_id: request.capability_snapshot_id, - observed_at_unix_ms: now, - max_evidence_age_ms: request.max_evidence_age_ms, - plan_version: request.plan_version, - activation_unix_ms: request.activation_unix_ms, - expiry_unix_ms: request.expiry_unix_ms, - backend_compat: request.backend_compat, - }, - ) { + let planning_request = physical::compiler::PlanningRequest { + queries, + evidence: request.evidence, + planner_revision: request.planner_revision, + }; + let environment = physical::compiler::DeploymentEnvironment { + target: physical::compiler::PhysicalDeploymentTarget::DistributedCollectors, + collector_ids: request.collector_ids.clone(), + capability_snapshot_id: request.capability_snapshot_id, + observed_at_unix_ms: now, + max_evidence_age_ms: request.max_evidence_age_ms, + plan_version: request.plan_version, + activation_unix_ms: request.activation_unix_ms, + expiry_unix_ms: request.expiry_unix_ms, + backend_compat: request.backend_compat, + }; + let candidates = physical::workload_cost::with_exact_alternative(planning_request.clone()) + .map_err(|error| (StatusCode::UNPROCESSABLE_ENTITY, error.to_string()))?; + let manifests = candidates + .iter() + .filter_map(|candidate| { + physical::compiler::PhysicalCompiler + .compile(candidate.clone(), environment.clone()) + .and_then(|plan| physical::workload_cost::manifest(&plan, &candidate.queries)) + .ok() + }) + .collect(); + let compiled = match request.workload_cost_evidence { + Some(evidence) => physical::workload_cost::select(candidates, environment, &evidence), + None => physical::compiler::PhysicalCompiler.compile(planning_request, environment), + }; + let bundle = match compiled { Ok(bundle) => bundle, Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), }; @@ -833,9 +856,27 @@ fn compile_physical_plan_request( request.collector_ids, apply_timeout, request.runtime_adaptation_evidence, + manifests, )) } +/// Read-only preparation: no OpAMP, staging, activation or data-plane writes. +async fn handle_workload_cost_manifests( + Json(request): Json, +) -> impl IntoResponse { + if request.workload_cost_evidence.is_some() { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + "omit quotes when requesting manifests", + ) + .into_response(); + } + match compile_physical_plan_request(request) { + Ok((_, _, _, _, manifests)) => Json(manifests).into_response(), + Err(error) => error.into_response(), + } +} + // ── Handlers ────────────────────────────────────────────────────────────────── async fn handle_plan(State(st): State, Json(spec): Json) -> impl IntoResponse { diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 929e20cd..563c7d6b 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -169,6 +169,8 @@ pub enum PhysicalDeploymentTarget { #[serde(deny_unknown_fields)] pub struct BackendLocalPlanningSnapshot { pub snapshot_version: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workload_cost_evidence: Option, pub query_workload: QueryWorkload, pub data_workload: DataWorkload, pub implementation: BackendLocalImplementation, @@ -620,6 +622,7 @@ pub struct PhysicalPlan { pub query_plan: QueryPlan, /// Lifecycle component only, not a complete physical-plan comparison. pub lifecycle_estimates: Vec, + pub cost_comparison: Option, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] @@ -1514,7 +1517,28 @@ impl BackendLocalPlanningSnapshot { /// one backend-local PhysicalPlan. No CollectorPlan is produced and no /// precompiled serving artifact is accepted at this boundary. pub fn compile(self) -> Result { - if self.snapshot_version != 1 { + let evidence = self.workload_cost_evidence.clone(); + if self.snapshot_version == 2 && evidence.is_none() { + return Err(CompileError::Snapshot( + "version 2 requires complete workload cost evidence".into(), + )); + } + let (request, environment) = self.planning_request()?; + match evidence { + Some(evidence) => super::workload_cost::select( + super::workload_cost::with_exact_alternative(request)?, + environment, + &evidence, + ), + None => PhysicalCompiler.compile(request, environment), + } + } + + /// Build Planner-authorized candidates for evidence collection without publishing. + pub fn planning_request( + self, + ) -> Result<(PlanningRequest, DeploymentEnvironment), CompileError> { + if self.snapshot_version != 1 && self.snapshot_version != 2 { return Err(CompileError::Snapshot(format!( "unsupported workload snapshot version {}", self.snapshot_version @@ -1633,14 +1657,14 @@ impl BackendLocalPlanningSnapshot { }); } select_workload_roots(&mut queries, canonical_roots, &HashMap::new())?; - PhysicalCompiler.compile( + Ok(( PlanningRequest { queries, evidence: HashMap::new(), planner_revision: PLANNER_REVISION.into(), }, self.environment, - ) + )) } } @@ -1693,6 +1717,17 @@ impl PhysicalCompiler { if let Some(e) = evidence { validate_evidence(&query.query_id, e, &environment)?; } + let node = query.post_asap.clone(); + let selected = + collect_selected_materializations(&node).map_err(|reason| CompileError::Query { + query_id: query.query_id.clone(), + reason, + })?; + // An exact native fallback has no maintained state and must not + // depend on evidence for unused window/state implementations. + if selected.is_empty() { + continue; + } validate_lifecycle_input(&query.query_id, &query.lifecycle)?; let lifecycle_costs = SummaryMaintenanceLifecycleCostInputs { build_cost: Some(Cost(query.lifecycle.costs.build)), @@ -1714,12 +1749,6 @@ impl PhysicalCompiler { }, ) .with_window_framework_costs(window_costs); - let node = query.post_asap.clone(); - let selected = - collect_selected_materializations(&node).map_err(|reason| CompileError::Query { - query_id: query.query_id.clone(), - reason, - })?; if environment.target == PhysicalDeploymentTarget::DistributedCollectors && selected.iter().any(|state| { matches!( @@ -1737,9 +1766,6 @@ impl PhysicalCompiler { .into(), }); } - if selected.is_empty() { - continue; - } match &query.source { Source::TimeSeries { .. } => {} Source::Table { .. } => { @@ -2040,6 +2066,7 @@ impl PhysicalCompiler { backend_plan, query_plan, lifecycle_estimates: lifecycle_estimates.into_values().collect(), + cost_comparison: None, }) } } @@ -3255,6 +3282,7 @@ mod tests { .remove(0); let snapshot = BackendLocalPlanningSnapshot { snapshot_version: 1, + workload_cost_evidence: None, query_workload, data_workload, implementation: BackendLocalImplementation { diff --git a/control_plane/src/physical/mod.rs b/control_plane/src/physical/mod.rs index a544efcf..e5d3c0c1 100644 --- a/control_plane/src/physical/mod.rs +++ b/control_plane/src/physical/mod.rs @@ -34,6 +34,7 @@ pub mod sketch_catalog; pub mod stage_split; pub mod topology; pub mod window_fusion; +pub mod workload_cost; pub mod workload_planner; // Convenience re-exports — preserve the surface that consumers of the diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs new file mode 100644 index 00000000..a89da2b8 --- /dev/null +++ b/control_plane/src/physical/workload_cost.rs @@ -0,0 +1,589 @@ +//! Complete, provider-priced comparisons of already-bound workload alternatives. +//! +//! This is an evidence manifest over the existing physical projection, not a +//! second semantic DAG. Planner supplies legal alternatives; deployment quotes +//! price every reachable operation, and the backend commits one complete plan. + +use std::collections::BTreeMap; + +use asap_aware_mapping::cost_model::Cost; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +use super::compiler::{ + CompileError, DeploymentEnvironment, PhysicalCompiler, PhysicalPlan, PlanningQuery, + PlanningRequest, +}; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct CostDemand { + /// Exact implementation/configuration being priced, not merely a family. + pub implementation: Value, + /// `horizon` includes all work in the manifest's source/time scope; + /// `query_evaluation` is one execution of this bound query operator. + pub unit: String, + pub multiplicity: f64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct WorkloadCostManifest { + pub plan_id: u64, + pub plan_version: u64, + pub planner_revision: String, + pub capability_snapshot_id: String, + pub backend_compat: String, + pub horizon_seconds: f64, + /// Canonical roots, requirements and demand must match across alternatives. + pub workload: BTreeMap, + pub components: BTreeMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct WorkloadQuote { + pub manifest: WorkloadCostManifest, + /// Provider attests feasibility for this capability/data generation, + /// including an accessible exact backend when the plan has fallback nodes. + pub executable: bool, + /// Calibrated costs in ONE model's units. All keys are required, including + /// explicit zero costs. Horizon quotes include source cardinality, all + /// maintained groups, retention/spill and the stated partition's work. + pub unit_costs: BTreeMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct WorkloadCostEvidence { + pub data_snapshot_id: String, + pub model_version: String, + pub observed_at_unix_ms: u64, + pub valid_for_ms: u64, + pub quotes: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct AlternativeCost { + pub plan_id: Option, + pub total_cost: Option, + pub unavailable_reason: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct WorkloadCostComparison { + pub data_snapshot_id: String, + pub model_version: String, + pub selected_plan_id: u64, + pub selected_manifest: WorkloadCostManifest, + pub component_costs: BTreeMap, + pub alternatives: Vec, +} + +fn invalid(reason: impl Into) -> CompileError { + CompileError::Snapshot(format!("complete workload cost: {}", reason.into())) +} + +pub fn manifest( + plan: &PhysicalPlan, + queries: &[PlanningQuery], +) -> Result { + let horizon = queries + .first() + .ok_or_else(|| invalid("empty workload"))? + .lifecycle + .horizon_seconds; + if !horizon.is_finite() || horizon <= 0.0 { + return Err(invalid("invalid horizon")); + } + let mut workload = BTreeMap::new(); + let mut reads = BTreeMap::new(); + for query in queries { + if query.lifecycle.horizon_seconds != horizon || query.lifecycle.evaluation_interval_ms == 0 + { + return Err(invalid("mixed horizons or unknown recurrence")); + } + let canonical = crate::query_plan::canonical_promql(&query.query_string)?; + if workload + .insert( + query.query_id.clone(), + json!({ + "query": canonical, "accuracy": query.accuracy, + "evaluation_interval_ms": query.lifecycle.evaluation_interval_ms, + "source": query.source, + }), + ) + .is_some() + { + return Err(invalid("duplicate query ID")); + } + reads.insert( + query.query_id.clone(), + horizon * 1000.0 / f64::from(query.lifecycle.evaluation_interval_ms), + ); + } + let mut components = BTreeMap::new(); + let mut add = |id: String, implementation: Value, unit: &str, multiplicity: f64| { + components.insert( + id, + CostDemand { + implementation, + unit: unit.into(), + multiplicity, + }, + ); + }; + // Backend merge/update, storage, and edge maintenance are separate work. + // The raw input is read once per source partition, not once per consumer. + for schema in &plan.precompute_plan.schemas { + let mut locations = vec!["backend".to_string()]; + locations.extend( + plan.collector_plans + .iter() + .filter(|collector| { + collector + .materializations + .iter() + .any(|m| m.materialization == schema.materialization) + }) + .map(|collector| format!("collector:{}", collector.collector_id)), + ); + for location in locations { + let source = json!({"source": schema.source, "location": location, "ingest": plan.precompute_plan.ingest}); + add(format!("source:{}", source), source, "horizon", 1.0); + let physical = plan + .precompute_plan + .materializations + .iter() + .find(|m| m.policy_fingerprint() == schema.materialization) + .ok_or_else(|| invalid("state has no physical implementation"))?; + let identity = json!({"schema": schema, "location": location, "physical": physical, + "window_implementation": plan.lifecycle_estimates.iter().find(|e| e.materialization == schema.materialization).map(|e| &e.window_implementation_id)}); + for operation in ["build", "update", "residency", "retire"] { + add( + format!("state:{location}:{}:{operation}", schema.materialization.0), + json!({"operation": operation, "binding": identity}), + "horizon", + 1.0, + ); + } + } + } + for rule in &plan.transmission_plan.rules { + add( + format!("transport:{}:{}", rule.producer_id, rule.materialization.0), + json!(rule), + "horizon", + 1.0, + ); + } + for entry in plan.query_plan.entries.values() { + let evaluations = *reads + .get(&entry.query_id) + .ok_or_else(|| invalid("unmapped query root"))?; + if entry + .nodes + .values() + .any(|node| matches!(node, crate::query_plan::QueryPlanNode::ExactFallback { .. })) + { + let query = queries + .iter() + .find(|query| query.query_id == entry.query_id) + .ok_or_else(|| invalid("unmapped exact source"))?; + // Charge the exact service's input upkeep/storage separately from + // per-evaluation native execution. Zero is valid only if explicitly + // quoted as already-provisioned/non-incremental for this decision. + let parsed = crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .map_err(|error| invalid(error.to_string()))?; + let metric = crate::query_parser::qe_to_parsed_query(&parsed).metric_name; + if metric.is_empty() { + return Err(invalid( + "exact-source pricing requires a unique time-series source in this profile", + )); + } + let source = json!({"source": planner_types::pre_asap::Source::TimeSeries { metric }, "location": "exact_backend"}); + add(format!("source:{}", source), source, "horizon", 1.0); + } + // Reachability comes from QueryPlan, including materialization reads, + // arithmetic, reduction and a complete engine-native exact fallback. + for node_id in entry.topological_order()? { + add( + format!("query:{}:{}", entry.query_id, node_id.0), + json!({"node": entry.nodes[&node_id], "query": entry.canonical_promql, "instant": entry.instant}), + "query_evaluation", + evaluations, + ); + } + add( + format!("result:{}", entry.query_id), + json!({"query_id": entry.query_id, "root": entry.root}), + "query_evaluation", + evaluations, + ); + } + Ok(WorkloadCostManifest { + plan_id: plan.envelope.plan_id, + plan_version: plan.envelope.plan_version, + planner_revision: plan.envelope.planner_revision.clone(), + capability_snapshot_id: plan.envelope.capability_snapshot_id.clone(), + backend_compat: plan.envelope.backend_compat.clone(), + horizon_seconds: horizon, + workload, + components, + }) +} + +impl WorkloadCostEvidence { + fn validate(&self, env: &DeploymentEnvironment) -> Result<(), CompileError> { + if self.data_snapshot_id.trim().is_empty() + || self.model_version.trim().is_empty() + || self.valid_for_ms == 0 + || self.observed_at_unix_ms > env.observed_at_unix_ms + || env.observed_at_unix_ms - self.observed_at_unix_ms + > self.valid_for_ms.min(env.max_evidence_age_ms) + { + return Err(invalid("missing, future or stale evidence generation")); + } + Ok(()) + } + + fn price( + &self, + manifest: &WorkloadCostManifest, + ) -> Result<(Cost, BTreeMap), String> { + let quotes = self + .quotes + .iter() + .filter(|quote| "e.manifest == manifest) + .collect::>(); + if quotes.len() != 1 { + return Err("missing or ambiguous quote for exact manifest".into()); + } + let quote = quotes[0]; + if !quote.executable { + return Err("provider reports unavailable implementation".into()); + } + if !quote.unit_costs.keys().eq(manifest.components.keys()) { + return Err("incomplete or extraneous component evidence".into()); + } + let mut total = 0.0; + let mut components = BTreeMap::new(); + for (id, demand) in &manifest.components { + let unit = quote.unit_costs[id]; + let cost = unit * demand.multiplicity; + if !unit.is_finite() || unit < 0.0 || !cost.is_finite() || cost < 0.0 { + return Err(format!("invalid cost for {id}")); + } + total += cost; + components.insert(id.clone(), cost); + } + if !total.is_finite() { + return Err("cost overflow".into()); + } + Ok((Cost(total), components)) + } +} + +/// Compare complete Planner-authorized forests after binding. Infeasible or +/// uncosted alternatives are retained as unavailable, never assigned zero. +pub fn select( + candidates: Vec, + env: DeploymentEnvironment, + evidence: &WorkloadCostEvidence, +) -> Result { + evidence.validate(&env)?; + if candidates.is_empty() || candidates.len() > 64 { + return Err(invalid( + "candidate inventory must contain 1..=64 alternatives", + )); + } + let mut comparison_workload = None; + let mut alternatives = Vec::new(); + let mut best: Option<( + Cost, + PhysicalPlan, + WorkloadCostManifest, + BTreeMap, + )> = None; + for candidate in candidates { + let queries = candidate.queries.clone(); + let plan = match PhysicalCompiler.compile(candidate, env.clone()) { + Ok(plan) => plan, + Err(error) => { + alternatives.push(AlternativeCost { + plan_id: None, + total_cost: None, + unavailable_reason: Some(error.to_string()), + }); + continue; + } + }; + let manifest = match manifest(&plan, &queries) { + Ok(manifest) => manifest, + Err(error) => { + alternatives.push(AlternativeCost { + plan_id: Some(plan.envelope.plan_id), + total_cost: None, + unavailable_reason: Some(error.to_string()), + }); + continue; + } + }; + let scope = (manifest.workload.clone(), manifest.horizon_seconds); + if comparison_workload + .as_ref() + .is_some_and(|previous| previous != &scope) + { + return Err(invalid( + "alternatives describe different workloads/horizons", + )); + } + comparison_workload = Some(scope); + match evidence.price(&manifest) { + Ok((cost, components)) => { + alternatives.push(AlternativeCost { + plan_id: Some(plan.envelope.plan_id), + total_cost: Some(cost.0), + unavailable_reason: None, + }); + if best.as_ref().is_none_or(|(previous, ..)| cost < *previous) { + best = Some((cost, plan, manifest, components)); + } + } + Err(reason) => alternatives.push(AlternativeCost { + plan_id: Some(plan.envelope.plan_id), + total_cost: None, + unavailable_reason: Some(reason), + }), + } + } + let (_, mut plan, selected_manifest, component_costs) = + best.ok_or_else(|| invalid("no feasible completely costed alternative"))?; + plan.cost_comparison = Some(WorkloadCostComparison { + data_snapshot_id: evidence.data_snapshot_id.clone(), + model_version: evidence.model_version.clone(), + selected_plan_id: plan.envelope.plan_id, + selected_manifest, + component_costs, + alternatives, + }); + Ok(plan) +} + +/// The current executor exposes continuously maintained state and the native +/// exact backend. Additional Planner-produced forests can use `select` directly. +pub fn with_exact_alternative( + request: PlanningRequest, +) -> Result, CompileError> { + let mut exact = request.clone(); + for query in &mut exact.queries { + let parsed = crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .map_err(|error| invalid(error.to_string()))?; + query.post_asap = crate::planner_selection::keep_pre_asap(&parsed) + .map_err(|error| invalid(error.to_string()))?; + } + if request + .queries + .iter() + .zip(&exact.queries) + .all(|(a, b)| a.post_asap == b.post_asap) + { + Ok(vec![request]) + } else { + Ok(vec![request, exact]) + } +} + +#[cfg(test)] +mod tests { + use super::super::compiler::BackendLocalPlanningSnapshot; + use super::*; + + fn fixture() -> BackendLocalPlanningSnapshot { + serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap() + } + + fn quoted() -> ( + Vec, + DeploymentEnvironment, + WorkloadCostEvidence, + ) { + let (request, env) = fixture().planning_request().unwrap(); + let candidates = with_exact_alternative(request).unwrap(); + let quotes = candidates + .iter() + .map(|candidate| { + let plan = PhysicalCompiler + .compile(candidate.clone(), env.clone()) + .unwrap(); + let manifest = manifest(&plan, &candidate.queries).unwrap(); + let unit_costs = manifest + .components + .keys() + .map(|id| (id.clone(), 1.0)) + .collect(); + WorkloadQuote { + manifest, + executable: true, + unit_costs, + } + }) + .collect(); + let evidence = WorkloadCostEvidence { + data_snapshot_id: "fixture-data-v1".into(), + model_version: "test-only-unit-costs".into(), + observed_at_unix_ms: env.observed_at_unix_ms, + valid_for_ms: env.max_evidence_age_ms, + quotes, + }; + (candidates, env, evidence) + } + + #[test] + fn complete_cost_changes_selection_and_reports_shared_work_once() { + let (candidates, env, mut evidence) = quoted(); + assert_ne!(evidence.quotes[0].manifest, evidence.quotes[1].manifest); + assert!(evidence.quotes[0] + .manifest + .components + .keys() + .any(|key| key.starts_with("state:"))); + assert!(!evidence.quotes[1] + .manifest + .components + .keys() + .any(|key| key.starts_with("state:"))); + for cost in evidence.quotes[1].unit_costs.values_mut() { + *cost = 1000.0; + } + let warm = select(candidates.clone(), env.clone(), &evidence).unwrap(); + assert_eq!(warm.envelope.plan_id, evidence.quotes[0].manifest.plan_id); + let report = warm.cost_comparison.unwrap(); + assert_eq!( + report.component_costs.len(), + report.selected_manifest.components.len() + ); + assert_eq!(report.alternatives.len(), 2); + assert!(report.alternatives.iter().all(|a| a.total_cost.is_some())); + for (id, cost) in &mut evidence.quotes[0].unit_costs { + if id.ends_with(":residency") { + *cost = 1e9; + } + } + let raw = select(candidates, env, &evidence).unwrap(); + assert_eq!(raw.envelope.plan_id, evidence.quotes[1].manifest.plan_id); + } + + #[test] + fn incomplete_unavailable_and_wrong_generation_quotes_are_not_free() { + let (candidates, env, mut evidence) = quoted(); + evidence.quotes[0].unit_costs.pop_first(); + let plan = select(candidates.clone(), env.clone(), &evidence).unwrap(); + assert!(plan.cost_comparison.unwrap().alternatives[0] + .unavailable_reason + .is_some()); + evidence.quotes[1].executable = false; + assert!(select(candidates.clone(), env.clone(), &evidence).is_err()); + let (_, _, mut evidence) = quoted(); + evidence + .quotes + .iter_mut() + .for_each(|quote| quote.manifest.capability_snapshot_id.push_str("-wrong")); + assert!(select(candidates.clone(), env.clone(), &evidence).is_err()); + let (_, _, mut evidence) = quoted(); + evidence.observed_at_unix_ms = env.observed_at_unix_ms + 1; + assert!(select(candidates, env, &evidence).is_err()); + } + + #[test] + fn v2_snapshot_requires_quotes_and_roundtrips_selection() { + let mut snapshot = fixture(); + snapshot.snapshot_version = 2; + assert!(snapshot.clone().compile().is_err()); + let (_, _, evidence) = quoted(); + snapshot.workload_cost_evidence = Some(evidence); + let snapshot: BackendLocalPlanningSnapshot = + serde_json::from_str(&serde_json::to_string(&snapshot).unwrap()).unwrap(); + assert!(snapshot.compile().unwrap().cost_comparison.is_some()); + } + + #[test] + fn second_consumer_adds_reads_not_another_shared_state() { + let (request, env) = fixture().planning_request().unwrap(); + let first = manifest( + &PhysicalCompiler + .compile(request.clone(), env.clone()) + .unwrap(), + &request.queries, + ) + .unwrap(); + let mut shared = request.clone(); + let mut second = shared.queries[0].clone(); + second.query_id = "second-consumer".into(); + second.query_string = "quantile_over_time(0.5, m[1m])".into(); + shared.queries.push(second); + let roots = shared + .queries + .iter() + .map(|query| { + std::rc::Rc::new( + crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .unwrap(), + ) + }) + .collect(); + super::super::compiler::select_workload_roots(&mut shared.queries, roots, &shared.evidence) + .unwrap(); + let plan = PhysicalCompiler.compile(shared.clone(), env).unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 1); + let second = manifest(&plan, &shared.queries).unwrap(); + let states = |m: &WorkloadCostManifest| { + m.components + .keys() + .filter(|key| key.starts_with("state:")) + .cloned() + .collect::>() + }; + assert_eq!(states(&first), states(&second)); + assert!(second.components.len() > first.components.len()); + } + + #[test] + fn exact_alternative_does_not_require_unused_state_implementation_evidence() { + let (candidates, env, evidence) = quoted(); + let mut exact = candidates[1].clone(); + assert_eq!(with_exact_alternative(exact.clone()).unwrap().len(), 1); + exact.queries[0].window_implementations.clear(); + assert!(select(vec![exact], env, &evidence).is_ok()); + } + + #[test] + fn altered_horizon_duplicate_and_invalid_costs_are_rejected() { + let (candidates, env, mut evidence) = quoted(); + evidence + .quotes + .iter_mut() + .for_each(|quote| quote.manifest.horizon_seconds += 1.0); + assert!(select(candidates.clone(), env.clone(), &evidence).is_err()); + let (_, _, mut evidence) = quoted(); + evidence.quotes.extend(evidence.quotes.clone()); + assert!(select(candidates.clone(), env.clone(), &evidence).is_err()); + let (_, _, mut evidence) = quoted(); + for quote in &mut evidence.quotes { + *quote.unit_costs.values_mut().next().unwrap() = -1.0; + } + assert!(select(candidates, env, &evidence).is_err()); + } +} diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index d85a1146..0ca072a2 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -484,6 +484,12 @@ async fn main() -> Result<()> { let plan = snapshot .compile() .map_err(|error| format!("startup planning failed for {}: {error}", path.display()))?; + if let Some(comparison) = &plan.cost_comparison { + info!( + "Startup workload cost decision: {}", + serde_json::to_string(comparison)? + ); + } Some( data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { precompute_plan: plan.precompute_plan, diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 02478793..249fa298 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -138,9 +138,46 @@ async fn shared_exact_dashboard_executes_selected_workload() { }) .collect(), ); - let typed: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + let mut typed: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(snapshot.clone()).unwrap(); + let (request, environment) = typed.clone().planning_request().unwrap(); + let candidates = + control_plane::physical::workload_cost::with_exact_alternative(request).unwrap(); + let quotes = candidates + .into_iter() + .enumerate() + .map(|(index, candidate)| { + let plan = control_plane::physical::compiler::PhysicalCompiler + .compile(candidate.clone(), environment.clone()) + .unwrap(); + let manifest = + control_plane::physical::workload_cost::manifest(&plan, &candidate.queries) + .unwrap(); + let unit_costs = manifest + .components + .keys() + .map(|key| (key.clone(), if index == 0 { 1.0 } else { 1000.0 })) + .collect(); + control_plane::physical::workload_cost::WorkloadQuote { + manifest, + executable: true, + unit_costs, + } + }) + .collect(); + typed.snapshot_version = 2; + typed.workload_cost_evidence = Some( + control_plane::physical::workload_cost::WorkloadCostEvidence { + data_snapshot_id: "process-fixture-v1".into(), + model_version: "test-only-unit-costs".into(), + observed_at_unix_ms: environment.observed_at_unix_ms, + valid_for_ms: environment.max_evidence_age_ms, + quotes, + }, + ); + snapshot = serde_json::to_value(&typed).unwrap(); let plan = typed.compile().unwrap(); + assert!(plan.cost_comparison.is_some()); assert_eq!(plan.precompute_plan.materializations.len(), 1); assert_eq!(plan.query_plan.entries.len(), 3); let snapshot_path = output_dir.path().join("snapshot.json"); diff --git a/data_plane/tests/backend_process_e2e.rs b/data_plane/tests/backend_process_e2e.rs index 2495614b..f6ed75ef 100644 --- a/data_plane/tests/backend_process_e2e.rs +++ b/data_plane/tests/backend_process_e2e.rs @@ -338,6 +338,60 @@ async fn respond_next_collector_plan( (plan, socket) } +async fn quote_workload( + client: &reqwest::Client, + control_base: &str, + request: &mut serde_json::Value, +) { + request + .as_object_mut() + .unwrap() + .remove("workload_cost_evidence"); + let manifests: Vec = client + .post(format!( + "{control_base}/api/v1/physical-plan/cost-manifests" + )) + .json(&request) + .send() + .await + .unwrap() + .error_for_status() + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(manifests.len(), 2); + let quotes = manifests + .into_iter() + .enumerate() + .map(|(index, manifest)| { + let unit_costs = manifest + .components + .keys() + .map(|key| (key.clone(), 1.0)) + .collect(); + control_plane::physical::workload_cost::WorkloadQuote { + manifest, + executable: index == 0, + unit_costs, + } + }) + .collect(); + request["workload_cost_evidence"] = serde_json::to_value( + control_plane::physical::workload_cost::WorkloadCostEvidence { + data_snapshot_id: "whole-process-fixture-v1".into(), + model_version: "test-only-unit-costs".into(), + observed_at_unix_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64, + valid_for_ms: 60000, + quotes, + }, + ) + .unwrap(); +} + #[tokio::test] async fn production_control_plane_to_data_plane_otlp_to_promql() { let control_binary = std::env::var("ASAP_E2E_CONTROL_PLANE_BIN") @@ -468,6 +522,10 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { second["query_id"] = "whole-process-e2e-median".into(); second["query_string"] = "quantile_over_time(0.5, whole_process_e2e_latency_ms[1s])".into(); request["queries"].as_array_mut().unwrap().push(second); + // Obtain actual compiler requirements without publishing, then provide + // deterministic test-only quotes. The native fallback is unavailable in + // this deployment; a cost number alone must not make it executable. + quote_workload(&client, &control_base, &mut request).await; let publication_response = client .post(format!( "{control_base}/api/v1/physical-plan/compile-and-publish" @@ -487,6 +545,14 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { ); let publication: serde_json::Value = serde_json::from_str(&publication_body).expect("decode publication response"); + assert_eq!( + publication["cost_comparison"]["alternatives"] + .as_array() + .unwrap() + .len(), + 2 + ); + assert!(publication["cost_comparison"]["alternatives"][1]["unavailable_reason"].is_string()); let (collector_plan, collector_socket) = collector.await.expect("collector task completed"); assert_eq!( publication["plan_id"], @@ -610,6 +676,7 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { .as_millis() as u64 + 1000) .into(); + quote_workload(&client, &control_base, &mut request).await; let collector = tokio::spawn(respond_next_collector_plan(collector_socket, false)); let failed = client .post(format!( diff --git a/docs/examples/workload-cost-evidence.md b/docs/examples/workload-cost-evidence.md new file mode 100644 index 00000000..bebde31a --- /dev/null +++ b/docs/examples/workload-cost-evidence.md @@ -0,0 +1,92 @@ +# Complete workload cost evidence + +The migrated, evidence-required startup profile uses `snapshot_version: 2`. +Version 1 and live requests without `workload_cost_evidence` remain compatibility +paths: their lifecycle estimates are not complete workload costs. + +## Workflow + +1. Prepare the canonical workload, deployment capabilities and implementation + evidence as in `asapquery-planning-snapshot.json`. Set version 2. +2. Obtain requirements without deploying: + + ```sh + cargo run -p control_plane --example workload_cost_manifest -- snapshot.json + ``` + + For distributed deployment, POST the normal compile-and-publish request, + without quotes, to `/api/v1/physical-plan/cost-manifests`. This endpoint only + compiles; it does not stage, publish to Collectors, or activate a generation. + It returns manifests for the compilable candidates in this runtime profile. +3. Have the deployment's cost/capability provider price each manifest. Attach a + `workload_cost_evidence` object to the snapshot or publication request: + + ```json + { + "data_snapshot_id": "deployment-input-generation-42", + "model_version": "calibrated-provider-v1", + "observed_at_unix_ms": 1780000000000, + "valid_for_ms": 60000, + "quotes": [{ + "manifest": "REPLACE with the complete returned manifest object", + "executable": true, + "unit_costs": { "REPLACE with every component ID": 1.0 } + }] + } + ``` + + This is a structural illustration, not runnable or calibrated evidence. + `unit_costs` must contain exactly the manifest's component keys. A quote + declares feasibility against the capability generation, including raw-data + access and the exact service when a candidate uses native fallback. +4. Compile/start or publish. The backend selects the cheapest completely quoted, + feasible candidate. The live response includes `cost_comparison`; startup + logs the same report. Preserve it with the deployment's evidence records. + +## What is priced + +This is a flat coverage manifest over the existing physical projection, not +another semantic DAG. Planner supplies the legal post-ASAP computations; costs +can change which bound workload is committed, not rewrite its semantics. + +- Source ingestion once per source/location over the horizon. +- Each shared state's build, update/merge, residency/spill and retirement once + per actual location, independent of the number of result consumers. +- Each Collector transmission rule over the horizon. +- Every reachable query operator, including reads, arithmetic, reductions and + complete engine-native fallback, multiplied by the query's recurrence. +- Each result's output cost. Exact-service input upkeep/storage has a separate + horizon component and is not silently omitted from a raw alternative. + +All quotes use one provider model's common cost units. A horizon quote includes +the complete stated partition's work, data volume/cardinality, maintained +groups and retention. Do not reuse a global ingestion rate as a per-metric rate. +Per-evaluation quotes exclude upkeep already charged in horizon components. +Sunk infrastructure may explicitly cost zero under the provider's documented +decision boundary; unknown costs may not. + +The manifest binds query requirements, demand, horizon, implementation details, +Planner revision, plan identity/version and capability generation. Evidence also +names its input generation, model and validity interval. A provider must refresh +quotes when those data assumptions change; this API does not discover or +authenticate telemetry itself. Freshness is evaluated at the supplied snapshot +decision time for replay, and at server time for live requests. + +Missing, duplicate, negative, nonfinite, stale, mismatched or infeasible quotes +make a candidate unavailable. If no completely quoted candidate remains, the +decision fails closed. Tests use explicit synthetic unit costs, not production +measurements. + +## Supported migration scope + +The default inventory compares the Planner-selected continuously maintained +workload with its whole-workload exact fallback. `workload_cost::select` also +accepts additional Planner-authorized, already-bindable forests. This does not +claim exhaustive search over every lifecycle, engine or Planner algorithm. +An exact alternative without an accessible native backend is unavailable even +if its numeric quote would be cheap. + +This provider-priced binding boundary deliberately avoids inventing physical +statistics to populate Planner's generic physical-formula provider. It completes +cost selection for the supported execution profile; production calibration and +additional provider implementations remain deployment work. From 5cc1d09dc92fa6df4716ee0575e0429c6f8e9acd Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 07:13:08 -0600 Subject: [PATCH 2/4] fix: enumerate feasible cost manifests without selecting a warm plan --- .../examples/workload_cost_manifest.rs | 8 +- control_plane/src/main.rs | 74 +++++++++++++++++-- 2 files changed, 73 insertions(+), 9 deletions(-) diff --git a/control_plane/examples/workload_cost_manifest.rs b/control_plane/examples/workload_cost_manifest.rs index e3463ce6..723fe8ed 100644 --- a/control_plane/examples/workload_cost_manifest.rs +++ b/control_plane/examples/workload_cost_manifest.rs @@ -12,13 +12,17 @@ fn main() -> Result<(), Box> { let (request, environment) = snapshot.planning_request()?; let manifests = workload_cost::with_exact_alternative(request)? .into_iter() - .map(|candidate| { + .filter_map(|candidate| { let queries = candidate.queries.clone(); PhysicalCompiler .compile(candidate, environment.clone()) .and_then(|plan| workload_cost::manifest(&plan, &queries)) + .ok() }) - .collect::, _>>()?; + .collect::>(); + if manifests.is_empty() { + return Err("no bindable workload cost manifests".into()); + } println!("{}", serde_json::to_string_pretty(&manifests)?); Ok(()) } diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index d2f37bf8..be6dc44d 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -637,8 +637,17 @@ async fn handle_compile_and_publish_physical_plan( Json(request): Json, ) -> impl IntoResponse { let (bundle, collector_ids, apply_timeout, adaptation_evidence, _) = - match compile_physical_plan_request(request) { - Ok(compiled) => compiled, + match compile_physical_plan_request(request, false) { + Ok((Some(bundle), ids, timeout, adaptation, manifests)) => { + (bundle, ids, timeout, adaptation, manifests) + } + Ok((None, ..)) => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "publication requires a selected plan", + ) + .into_response() + } Err(response) => return response.into_response(), }; @@ -733,9 +742,10 @@ async fn handle_compile_and_publish_physical_plan( // Only the Send-safe compiled bundle crosses an await point. fn compile_physical_plan_request( request: CompileAndPublishPhysicalPlanRequest, + manifests_only: bool, ) -> Result< ( - physical::compiler::PhysicalPlan, + Option, Vec, Duration, Vec, @@ -833,7 +843,7 @@ fn compile_physical_plan_request( }; let candidates = physical::workload_cost::with_exact_alternative(planning_request.clone()) .map_err(|error| (StatusCode::UNPROCESSABLE_ENTITY, error.to_string()))?; - let manifests = candidates + let manifests: Vec<_> = candidates .iter() .filter_map(|candidate| { physical::compiler::PhysicalCompiler @@ -842,6 +852,24 @@ fn compile_physical_plan_request( .ok() }) .collect(); + let apply_timeout = Duration::from_millis(request.apply_timeout_ms); + // Quote preparation enumerates feasible bindings; it does not select the + // default warm candidate, which may be unavailable while exact is valid. + if manifests_only { + if manifests.is_empty() { + return Err(( + StatusCode::UNPROCESSABLE_ENTITY, + "no bindable workload cost manifests".into(), + )); + } + return Ok(( + None, + request.collector_ids, + apply_timeout, + request.runtime_adaptation_evidence, + manifests, + )); + } let compiled = match request.workload_cost_evidence { Some(evidence) => physical::workload_cost::select(candidates, environment, &evidence), None => physical::compiler::PhysicalCompiler.compile(planning_request, environment), @@ -850,9 +878,8 @@ fn compile_physical_plan_request( Ok(bundle) => bundle, Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), }; - let apply_timeout = Duration::from_millis(request.apply_timeout_ms); Ok(( - bundle, + Some(bundle), request.collector_ids, apply_timeout, request.runtime_adaptation_evidence, @@ -871,7 +898,7 @@ async fn handle_workload_cost_manifests( ) .into_response(); } - match compile_physical_plan_request(request) { + match compile_physical_plan_request(request, true) { Ok((_, _, _, _, manifests)) => Json(manifests).into_response(), Err(error) => error.into_response(), } @@ -2082,6 +2109,39 @@ mod api_tests { use http_body_util::BodyExt; use tower::ServiceExt; + // A missing warm implementation must not hide the executable exact quote. + #[tokio::test] + async fn cost_manifests_survive_unavailable_warm_candidate() { + let snapshot: physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_str( + include_str!("../../docs/examples/asapquery-planning-snapshot.json"), + ) + .unwrap(); + let (planning, _) = snapshot.planning_request().unwrap(); + let query = &planning.queries[0]; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + let request = serde_json::from_value(serde_json::json!({ + "queries": [{ + "query_id": query.query_id, "query_string": query.query_string, + "metric": "m", "window_secs": 60, "accuracy": query.accuracy, + "lifecycle": query.lifecycle, "window_implementations": [] + }], + "collector_ids": ["test"], "capability_snapshot_id": "test", + "planner_revision": physical::compiler::PLANNER_REVISION, + "max_evidence_age_ms": 60000, "plan_version": 1, + "activation_unix_ms": now, "backend_compat": control_plane::backend_plan::BACKEND_COMPAT + })) + .unwrap(); + let response = handle_workload_cost_manifests(Json(request)) + .await + .into_response(); + assert_eq!(response.status(), StatusCode::OK); + let manifests = body_json(response).await; + assert_eq!(manifests.as_array().unwrap().len(), 1); + } + async fn body_json(resp: axum::response::Response) -> serde_json::Value { let bytes = resp.into_body().collect().await.unwrap().to_bytes(); serde_json::from_slice(&bytes).unwrap() From 558080765bf3706ee936f93af713b90f11d0aeac Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 12:45:07 -0600 Subject: [PATCH 3/4] fix(cost): price every exact fallback input source --- control_plane/src/physical/workload_cost.rs | 142 ++++++++++++++++++-- docs/examples/workload-cost-evidence.md | 5 +- 2 files changed, 138 insertions(+), 9 deletions(-) diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index a89da2b8..d2baa296 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -4,7 +4,7 @@ //! second semantic DAG. Planner supplies legal alternatives; deployment quotes //! price every reachable operation, and the backend commits one complete plan. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use asap_aware_mapping::cost_model::Cost; use serde::{Deserialize, Serialize}; @@ -198,14 +198,10 @@ pub fn manifest( query.accuracy.clone(), ) .map_err(|error| invalid(error.to_string()))?; - let metric = crate::query_parser::qe_to_parsed_query(&parsed).metric_name; - if metric.is_empty() { - return Err(invalid( - "exact-source pricing requires a unique time-series source in this profile", - )); + for metric in exact_source_metrics(&parsed)? { + let source = json!({"source": planner_types::pre_asap::Source::TimeSeries { metric }, "location": "exact_backend"}); + add(format!("source:{}", source), source, "horizon", 1.0); } - let source = json!({"source": planner_types::pre_asap::Source::TimeSeries { metric }, "location": "exact_backend"}); - add(format!("source:{}", source), source, "horizon", 1.0); } // Reachability comes from QueryPlan, including materialization reads, // arithmetic, reduction and a complete engine-native exact fallback. @@ -236,6 +232,65 @@ pub fn manifest( }) } +/// Walk the canonical relational tree, preserving every input to binary and +/// fan-in operators. Unsupported source discovery must not produce a partial quote. +fn exact_source_metrics( + expr: &planner_types::pre_asap::QueryExpr, +) -> Result, CompileError> { + use planner_types::pre_asap::{QueryExpr, Source}; + fn visit(expr: &QueryExpr, metrics: &mut BTreeSet) -> Result<(), CompileError> { + match expr { + QueryExpr::Scan { + source: Source::TimeSeries { metric }, + .. + } if !metric.is_empty() => { + metrics.insert(metric.clone()); + } + QueryExpr::PromqlScalarBridge(child) + | QueryExpr::PromqlVectorFromScalar(child) + | QueryExpr::PromqlScalarFromVector(child) + | QueryExpr::PromqlRelabel { child, .. } + | QueryExpr::PromqlSeriesSample { child, .. } + | QueryExpr::Filter { child, .. } + | QueryExpr::Project { child, .. } + | QueryExpr::Aggregate { child, .. } + | QueryExpr::Dedup { child, .. } + | QueryExpr::Sort { child, .. } + | QueryExpr::Limit { child, .. } + | QueryExpr::PromqlSubquery { child, .. } + | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } => visit(child, metrics)?, + QueryExpr::BinaryOp { + lhs: left, + rhs: right, + .. + } + | QueryExpr::Join { left, right, .. } + | QueryExpr::SetOp { left, right, .. } => { + visit(left, metrics)?; + visit(right, metrics)?; + } + QueryExpr::Concat { children, .. } => { + for child in children { + visit(child, metrics)?; + } + } + QueryExpr::Literal(_) | QueryExpr::EvalTimestamp => {} + // In particular, info() has an implicit metadata source that is + // not a Scan child, and unnamed selectors require source discovery. + _ => { + return Err(invalid( + "exact-source pricing cannot enumerate this query's sources", + )) + } + } + Ok(()) + } + let mut metrics = BTreeSet::new(); + visit(expr, &mut metrics)?; + Ok(metrics) +} + impl WorkloadCostEvidence { fn validate(&self, env: &DeploymentEnvironment) -> Result<(), CompileError> { if self.data_snapshot_id.trim().is_empty() @@ -448,6 +503,77 @@ mod tests { (candidates, env, evidence) } + // All input metrics need upkeep quotes; repeated reads share that upkeep. + #[test] + fn exact_manifest_covers_and_deduplicates_query_sources() { + for (query, expected) in [ + ( + "sum_over_time(m[1m]) + sum_over_time(n[1m])", + vec!["m", "n"], + ), + ("sum_over_time(m[1m]) + count_over_time(m[1m])", vec!["m"]), + ] { + let (mut request, env) = fixture().planning_request().unwrap(); + request.queries[0].query_string = query.into(); + let exact = with_exact_alternative(request).unwrap().pop().unwrap(); + let plan = PhysicalCompiler + .compile(exact.clone(), env.clone()) + .unwrap(); + let manifest = manifest(&plan, &exact.queries).unwrap(); + let sources: Vec<_> = manifest + .components + .iter() + .filter(|(id, _)| id.starts_with("source:")) + .map(|(_, demand)| { + demand.implementation["source"]["TimeSeries"]["metric"] + .as_str() + .unwrap() + }) + .collect(); + assert_eq!(sources, expected, "{query}"); + let mut evidence = WorkloadCostEvidence { + data_snapshot_id: "test-data".into(), + model_version: "test-model".into(), + observed_at_unix_ms: env.observed_at_unix_ms, + valid_for_ms: env.max_evidence_age_ms, + quotes: vec![WorkloadQuote { + unit_costs: manifest + .components + .keys() + .map(|id| (id.clone(), 1.0)) + .collect(), + manifest, + executable: true, + }], + }; + assert!(select(vec![exact.clone()], env.clone(), &evidence).is_ok()); + let source_id = evidence.quotes[0] + .unit_costs + .keys() + .rfind(|id| id.starts_with("source:")) + .unwrap() + .clone(); + evidence.quotes[0].unit_costs.remove(&source_id); + assert!( + select(vec![exact], env, &evidence).is_err(), + "missing input upkeep must fail closed" + ); + } + } + + // Hidden or unresolved sources must not yield a partially priced manifest. + #[test] + fn exact_source_discovery_rejects_unresolved_inputs() { + let accuracy = fixture().planning_request().unwrap().0.queries[0] + .accuracy + .clone(); + for query in ["info(m)", "{job=\"api\"}"] { + let parsed = + crate::query_parser::parse_query_expr_canonical(query, accuracy.clone()).unwrap(); + assert!(exact_source_metrics(&parsed).is_err(), "{query}"); + } + } + #[test] fn complete_cost_changes_selection_and_reports_shared_work_once() { let (candidates, env, mut evidence) = quoted(); diff --git a/docs/examples/workload-cost-evidence.md b/docs/examples/workload-cost-evidence.md index bebde31a..612f58e7 100644 --- a/docs/examples/workload-cost-evidence.md +++ b/docs/examples/workload-cost-evidence.md @@ -49,7 +49,10 @@ This is a flat coverage manifest over the existing physical projection, not another semantic DAG. Planner supplies the legal post-ASAP computations; costs can change which bound workload is committed, not rewrite its semantics. -- Source ingestion once per source/location over the horizon. +- Source ingestion once per source/location over the horizon. Exact fallback + includes every named input metric, including both sides of binary expressions; + repeated references to a metric share one upkeep component. Queries whose + sources cannot be fully enumerated are unavailable for complete costing. - Each shared state's build, update/merge, residency/spill and retirement once per actual location, independent of the number of result consumers. - Each Collector transmission rule over the horizon. From 9beb9f6a30dd0a5dd366fdcc24c76dfcbac7e18f Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 12:50:21 -0600 Subject: [PATCH 4/4] build: pin Planner to merged shared-resource and offline-evidence APIs --- Cargo.lock | 6 +++--- control_plane/Cargo.toml | 6 +++--- control_plane/src/physical/compiler.rs | 2 +- crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 2 +- 5 files changed, 9 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9eb7cadb..5f1fb4b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -343,7 +343,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=378a7547ede629a64e84c9f7c810226ce196cce9#378a7547ede629a64e84c9f7c810226ce196cce9" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=c3410d14865497758d212e4265ad25c782187de1#c3410d14865497758d212e4265ad25c782187de1" dependencies = [ "asap-types", "serde", @@ -354,7 +354,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=378a7547ede629a64e84c9f7c810226ce196cce9#378a7547ede629a64e84c9f7c810226ce196cce9" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=c3410d14865497758d212e4265ad25c782187de1#c3410d14865497758d212e4265ad25c782187de1" dependencies = [ "asap-types", "promql-parser 0.10.0", @@ -374,7 +374,7 @@ dependencies = [ [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=378a7547ede629a64e84c9f7c810226ce196cce9#378a7547ede629a64e84c9f7c810226ce196cce9" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=c3410d14865497758d212e4265ad25c782187de1#c3410d14865497758d212e4265ad25c782187de1" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 0c276d99..52d7961c 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -93,8 +93,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -102,7 +102,7 @@ asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index ad49df39..8079714a 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -42,7 +42,7 @@ use crate::query_plan::{ use crate::types_v2::AccuracyTarget; use planner_types::pre_asap::Source; -pub const PLANNER_REVISION: &str = "378a7547ede629a64e84c9f7c810226ce196cce9"; +pub const PLANNER_REVISION: &str = "c3410d14865497758d212e4265ad25c782187de1"; #[derive(Debug, Clone)] pub struct PlanningQuery { diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index 2d520078..d093312d 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -30,4 +30,4 @@ xxhash-rust = { version = "0.8", features = ["xxh64"] } # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 50b3a0d9..f8b43fbe 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -38,7 +38,7 @@ control_plane = { path = "../control_plane" } # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } # Shared external (workspace) serde.workspace = true