From 577bd83e3f4fb1ac91d5a964e904a78371148df3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 11:22:11 -0600 Subject: [PATCH 1/2] Require complete cost evidence for snapshot deployment --- control_plane/src/backend_client.rs | 6 +- control_plane/src/physical/compiler.rs | 190 +++++++++++++----- control_plane/src/physical/erp.rs | 4 +- control_plane/src/physical/workload_cost.rs | 3 +- .../asapquery_compatibility_process_e2e.rs | 69 ++++++- .../support/distinct_planning_process.rs | 21 +- .../tests/support/durable_summary_process.rs | 2 +- .../tests/support/erp_planning_process.rs | 5 +- .../support/immutable_maintenance_process.rs | 17 +- .../tests/support/univmon_erp_process.rs | 16 +- ...asapquery-compatibility-demo-snapshot.json | 2 +- .../examples/asapquery-planning-snapshot.json | 2 +- 12 files changed, 248 insertions(+), 89 deletions(-) diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index fb60d3cc..75841913 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -601,7 +601,11 @@ mod tests { "../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - let publication = snapshot.compile().unwrap().publication().unwrap(); + let publication = crate::physical::compiler::tests::quoted_snapshot(snapshot, false) + .compile() + .unwrap() + .publication() + .unwrap(); let hits: StdArc>> = StdArc::new(Mutex::new(Vec::new())); let route_hits = hits.clone(); let app = Router::new().route( diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index b57f00e8..3528a6a1 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -198,7 +198,8 @@ pub enum PhysicalDeploymentTarget { BackendLocalRemoteWrite, } -/// Versioned startup input for the Collector-free compatibility profile. +/// Startup and candidate-discovery input for backend-local planning. +/// Version 2 is the sole supported schema; deployment always requires quotes. /// Query/data semantics use ASAPPlanner's canonical workload types directly; /// this wrapper adds only backend-owned implementation evidence and lifecycle /// identity required to choose a concrete physical realization. @@ -206,6 +207,7 @@ pub enum PhysicalDeploymentTarget { #[serde(deny_unknown_fields)] pub struct BackendLocalPlanningSnapshot { pub snapshot_version: u32, + /// May be absent during candidate discovery, never during deployment. #[serde(default, skip_serializing_if = "Option::is_none")] pub workload_cost_evidence: Option, pub query_workload: QueryWorkload, @@ -480,34 +482,17 @@ impl BackendLocalPlanningSnapshot { } fn compile_frontend(self, metricsql: bool) -> Result { - 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 evidence = self.workload_cost_evidence.clone().ok_or_else(|| { + CompileError::Snapshot( + "deployment requires complete workload cost evidence; export candidates and price them before compiling".into(), + ) + })?; let (request, environment) = self.planning_request()?; - match evidence { - Some(evidence) => { - let candidates = super::workload_cost::with_exact_alternative(request)?; - if metricsql { - super::workload_cost::select_metricsql(candidates, environment, &evidence) - } else { - super::workload_cost::select(candidates, environment, &evidence) - } - } - None => { - // Unquoted v1 startup snapshots keep the established summary/native - // compatibility policy. Local residual candidates are enumerated by - // planning_request and admitted through measured workload selection. - let mut request = request; - request.hybrid_execution = false; - if metricsql { - PhysicalCompiler.compile_metricsql(request, environment) - } else { - PhysicalCompiler.compile(request, environment) - } - } + let candidates = super::workload_cost::with_exact_alternative(request)?; + if metricsql { + super::workload_cost::select_metricsql(candidates, environment, &evidence) + } else { + super::workload_cost::select(candidates, environment, &evidence) } } @@ -515,9 +500,9 @@ impl BackendLocalPlanningSnapshot { pub fn planning_request( self, ) -> Result<(PlanningRequest, DeploymentEnvironment), CompileError> { - if self.snapshot_version != 1 && self.snapshot_version != 2 { + if self.snapshot_version != 2 { return Err(CompileError::Snapshot(format!( - "unsupported workload snapshot version {}", + "unsupported workload snapshot version {}; only version 2 is supported", self.snapshot_version ))); } @@ -1058,7 +1043,9 @@ impl PhysicalCompiler { .ok() .flatten() }); - key.is_some_and(|key| policy.contains(&key)) + // Masks enumerate counter/max choices only. Other selected + // summaries remain required by this physical alternative. + key.is_none_or(|key| policy.contains(&key)) }) }) .collect::>(); @@ -1553,7 +1540,31 @@ impl PhysicalCompiler { full_history: false, cumulative_readout: true, }; - let mut entry = if request.hybrid_execution { + // A whole-query native fallback need not be expressible in the local + // residual algebra (for example an ERP-rejected entropy readout). + // Retain its native boundary without discarding other workload roots. + let native_root = request.hybrid_execution + && if let SummaryExpr::KeepPreAsap(expr) = &query.post_asap.expr { + let original = crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .map_err(|error| CompileError::Query { + query_id: query.query_id.clone(), + reason: error.to_string(), + })?; + expr.as_ref() == &original + && crate::query_plan::logical::compile_logical( + query.query_id.clone(), + canonical.clone(), + instant.clone(), + FallbackPolicy::ExactBackend, + ) + .is_err() + } else { + false + }; + let mut entry = if request.hybrid_execution && !native_root { crate::query_plan::compile_bound_composable_mapped( query.query_id.clone(), canonical.clone(), @@ -3237,9 +3248,88 @@ fn stable_workload_plan_id( } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; + // Synthetic quotes exercise deployment selection in tests, never production defaults. + pub(crate) fn quoted_snapshot( + mut snapshot: BackendLocalPlanningSnapshot, + metricsql: bool, + ) -> BackendLocalPlanningSnapshot { + use super::super::workload_cost::{ + manifest, with_exact_alternative, WorkloadCostEvidence, WorkloadQuote, + }; + let (request, environment) = snapshot.clone().planning_request().unwrap(); + let quotes = with_exact_alternative(request) + .unwrap() + .into_iter() + .enumerate() + .filter_map(|(index, candidate)| { + let plan = if metricsql { + PhysicalCompiler.compile_metricsql(candidate.clone(), environment.clone()) + } else { + PhysicalCompiler.compile(candidate.clone(), environment.clone()) + } + .ok()?; + let manifest = manifest(&plan, &candidate.queries).unwrap(); + Some(WorkloadQuote { + unit_costs: manifest + .components + .keys() + .map(|key| (key.clone(), if index == 0 { 1.0 } else { 1e12 })) + .collect(), + manifest, + executable: true, + }) + }) + .collect(); + snapshot.workload_cost_evidence = Some(WorkloadCostEvidence { + backend_revision: BACKEND_REVISION.into(), + planner_revision: PLANNER_REVISION.into(), + data_snapshot_id: "compiler-unit-fixture".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 + } + + /// Optional counter masks must retain the workload's mandatory sketch bindings. + #[test] + fn costed_mixed_workload_retains_sketches_and_counter_readouts() { + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let plan = quoted_snapshot(snapshot, false).compile().unwrap(); + assert!(!plan.precompute_plan.materializations.is_empty()); + for entry in plan.query_plan.entries.values() { + assert!(!entry.materialization_bindings().is_empty(), "{entry:#?}"); + } + } + + /// A schema marker cannot opt into a legacy deployment policy. + #[test] + fn only_current_snapshot_schema_is_accepted() { + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + for version in [0, 1, 3] { + let mut old = snapshot.clone(); + old.snapshot_version = version; + assert!(old + .clone() + .planning_request() + .unwrap_err() + .to_string() + .contains("only version 2")); + assert!(old.compile().is_err()); + } + assert!(snapshot.planning_request().is_ok()); + } + #[test] fn installed_partition_must_match_the_bound_dag_reduction() { let mut env = environment(10_000); @@ -3957,7 +4047,7 @@ mod tests { "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile_metricsql().unwrap(); + let plan = quoted_snapshot(snapshot, true).compile_metricsql().unwrap(); assert!(!plan.query_plan.entries.is_empty()); assert!(plan .query_plan @@ -4993,7 +5083,7 @@ mod tests { "../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - let bundle = snapshot.compile().unwrap(); + let bundle = quoted_snapshot(snapshot, false).compile().unwrap(); let catalog = &bundle.summary_catalog; let mut transmission = bundle.transmission_plan.clone(); transmission.validate_against_catalog(catalog).unwrap(); @@ -5061,7 +5151,7 @@ mod tests { let mut second = entries[0].clone(); second.query = Query("sum(sum_over_time(m[1m])) * 2".into()); entries.push(second); - let bundle = snapshot.compile().unwrap(); + let bundle = quoted_snapshot(snapshot, false).compile().unwrap(); assert_eq!(bundle.query_plan.entries.len(), 2); assert_eq!(bundle.precompute_plan.materializations.len(), 1); let query_plan: QueryPlan = @@ -5403,7 +5493,7 @@ mod tests { .queries .remove(0); let snapshot = BackendLocalPlanningSnapshot { - snapshot_version: 1, + snapshot_version: 2, workload_cost_evidence: None, query_workload, data_workload, @@ -5434,12 +5524,10 @@ mod tests { .as_ref(), Some(&snapshot.query_workload) ); - let first = snapshot - .clone() + let first = quoted_snapshot(snapshot.clone(), false) .compile() .expect("first deterministic plan"); - let second = snapshot - .clone() + let second = quoted_snapshot(snapshot.clone(), false) .compile() .expect("second deterministic plan"); assert_eq!(first.envelope, second.envelope); @@ -5454,7 +5542,9 @@ mod tests { let encoded = serde_json::to_vec(&snapshot).expect("serialize startup snapshot"); let decoded: BackendLocalPlanningSnapshot = serde_json::from_slice(&encoded).expect("deserialize startup snapshot"); - let bundle = decoded.compile().expect("canonical startup planning"); + let bundle = quoted_snapshot(decoded, false) + .compile() + .expect("canonical startup planning"); assert!(bundle.collector_plans.is_empty()); assert!(bundle.transmission_plan.rules.is_empty()); @@ -5542,10 +5632,10 @@ mod tests { let fixture: serde_json::Value = serde_json::from_str(source).expect("fixture JSON"); assert_eq!(encoded, fixture); - snapshot - .clone() - .compile() - .expect("unquoted v1 compatibility startup remains available"); + assert!( + snapshot.clone().compile().is_err(), + "discovery fixtures must be priced before deployment" + ); let (local, env) = snapshot.clone().planning_request().unwrap(); let isolated = PhysicalCompiler.compile(local, env).unwrap(); assert!(!isolated.precompute_plan.materializations.is_empty()); @@ -5578,10 +5668,10 @@ mod tests { include_str!("../../../docs/examples/asapquery-compatibility-demo-snapshot.json"); let snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(source).expect("strict compatibility demo fixture"); - snapshot - .clone() - .compile() - .expect("unquoted v1 compatibility startup remains available"); + assert!( + snapshot.clone().compile().is_err(), + "discovery fixtures must be priced before deployment" + ); let (local, env) = snapshot.clone().planning_request().unwrap(); let isolated = PhysicalCompiler.compile(local, env).unwrap(); assert!(!isolated.precompute_plan.materializations.is_empty()); diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index c41fa2fe..a9e139f1 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -1459,7 +1459,9 @@ mod tests { fixture["query_workload"]["repeating_queries"] = serde_json::json!([query]); let snapshot: crate::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::physical::compiler::tests::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let (mut policy, mut observed) = online_population_fixture(); observed.catalog_generation = plan.summary_catalog.reference().unwrap(); observed.summary_definition_id = diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index ecf98038..b3dde46c 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -1144,9 +1144,8 @@ mod tests { } #[test] - fn v2_snapshot_requires_quotes_and_roundtrips_selection() { + fn 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); diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 54cad0d8..e6953605 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -28,6 +28,60 @@ mod durable_summary_process; #[path = "support/immutable_maintenance_process.rs"] mod immutable_maintenance_process; +// Test-only quotes preserve the fixture's local candidate without a production bypass. +fn quote_snapshot_for_test( + snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot, +) -> control_plane::physical::compiler::BackendLocalPlanningSnapshot { + quote_snapshot_for_frontend_test(snapshot, false) +} + +fn quote_snapshot_for_frontend_test( + mut snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot, + metricsql: bool, +) -> control_plane::physical::compiler::BackendLocalPlanningSnapshot { + use control_plane::physical::{ + compiler::{PhysicalCompiler, BACKEND_REVISION, PLANNER_REVISION}, + workload_cost::{self, WorkloadCostEvidence, WorkloadQuote}, + }; + let (request, environment) = snapshot.clone().planning_request().unwrap(); + let mut preferred = true; + let quotes = workload_cost::with_exact_alternative(request) + .unwrap() + .into_iter() + .enumerate() + .filter_map(|(_index, candidate)| { + let plan = if metricsql { + PhysicalCompiler.compile_metricsql(candidate.clone(), environment.clone()) + } else { + PhysicalCompiler.compile(candidate.clone(), environment.clone()) + } + .ok()?; + let unit_cost = if preferred { 1.0 } else { 1e12 }; + preferred = false; + let manifest = workload_cost::manifest(&plan, &candidate.queries).unwrap(); + Some(WorkloadQuote { + unit_costs: manifest + .components + .keys() + .map(|key| (key.clone(), unit_cost)) + .collect(), + manifest, + executable: true, + }) + }) + .collect(); + snapshot.workload_cost_evidence = Some(WorkloadCostEvidence { + backend_revision: BACKEND_REVISION.into(), + planner_revision: PLANNER_REVISION.into(), + data_snapshot_id: "process-fixture".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 +} + struct ChildGuard(Child); impl Drop for ChildGuard { @@ -1143,15 +1197,18 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() let backend_port = unused_port(); let output_dir = tempfile::tempdir().expect("backend output directory"); - let snapshot = concat!( - env!("CARGO_MANIFEST_DIR"), - "/../docs/examples/asapquery-compatibility-demo-snapshot.json" - ); + let fixture = serde_json::from_str(include_str!( + "../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let priced = quote_snapshot_for_test(fixture); + let snapshot = output_dir.path().join("snapshot.json"); + std::fs::write(&snapshot, serde_json::to_vec(&priced).unwrap()).unwrap(); let child = Command::new(env!("CARGO_BIN_EXE_data_plane")) .arg("--profile") .arg("asapquery") .arg("--planning-snapshot") - .arg(snapshot) + .arg(&snapshot) .arg("--prometheus-server") .arg(format!("http://{fallback_address}")) .arg("--forward-unsupported-queries") @@ -1588,7 +1645,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() .as_array() .expect("materialization statuses"); let planned_snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = - serde_json::from_str(&std::fs::read_to_string(snapshot).unwrap()).unwrap(); + serde_json::from_str(&std::fs::read_to_string(&snapshot).unwrap()).unwrap(); let planned = planned_snapshot.compile().unwrap(); // Every selected state must be serving; the Planner may share or separate // physical populations, so compare identities rather than a frozen count. diff --git a/data_plane/tests/support/distinct_planning_process.rs b/data_plane/tests/support/distinct_planning_process.rs index 123624db..a67a197a 100644 --- a/data_plane/tests/support/distinct_planning_process.rs +++ b/data_plane/tests/support/distinct_planning_process.rs @@ -23,10 +23,11 @@ async fn distinct_range_uses_planner_selected_hll_and_source_labels() { entry["query"] = QUERY.into(); entry["requirements"]["accuracy"] = serde_json::json!({"explicit": {"Epsilon": 0.05}}); fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); - let plan = serde_json::from_value::(fixture.clone()) - .unwrap() - .compile() - .unwrap(); + let plan = quote_snapshot_for_test( + serde_json::from_value::(fixture.clone()).unwrap(), + ) + .compile() + .unwrap(); assert_eq!(plan.precompute_plan.materializations.len(), 1); assert_eq!( plan.precompute_plan.materializations[0].aggregation_type, @@ -38,7 +39,8 @@ async fn distinct_range_uses_planner_selected_hll_and_source_labels() { ); let output = tempfile::tempdir().unwrap(); let path = output.path().join("planning.json"); - std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap(); + let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap()); + std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap(); let port = unused_port(); let mut vm_port = unused_port(); while vm_port == port { @@ -80,11 +82,10 @@ async fn distinct_range_uses_planner_selected_hll_and_source_labels() { // Source syntax uses the shared parser fork; serving semantics and exact // routing belong to the MetricsQL adapter and its installed query entries. let snapshot = serde_json::from_value::(fixture).unwrap(); - let (mut request, mut environment) = snapshot.planning_request().unwrap(); - request.hybrid_execution = false; - environment.plan_version = 2; - let compiled = control_plane::physical::compiler::PhysicalCompiler - .compile_metricsql(request, environment) + let mut snapshot = snapshot; + snapshot.environment.plan_version = 2; + let compiled = quote_snapshot_for_frontend_test(snapshot, true) + .compile_metricsql() .unwrap(); let identity = serde_json::json!({"plan_id": compiled.envelope.plan_id, "plan_version": compiled.envelope.plan_version}); let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { diff --git a/data_plane/tests/support/durable_summary_process.rs b/data_plane/tests/support/durable_summary_process.rs index b37d39c8..626392ae 100644 --- a/data_plane/tests/support/durable_summary_process.rs +++ b/data_plane/tests/support/durable_summary_process.rs @@ -15,7 +15,7 @@ async fn persisted_summary_restarts_without_live_reregistration() { serde_json::json!([fixture["query_workload"]["repeating_queries"][2].clone()]); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = quote_snapshot_for_test(snapshot).compile().unwrap(); let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { summary_catalog: plan.summary_catalog, collector_plans: plan.collector_plans, diff --git a/data_plane/tests/support/erp_planning_process.rs b/data_plane/tests/support/erp_planning_process.rs index aa0cf508..6e88beac 100644 --- a/data_plane/tests/support/erp_planning_process.rs +++ b/data_plane/tests/support/erp_planning_process.rs @@ -104,7 +104,7 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write() )); let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture.clone()).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = quote_snapshot_for_test(snapshot).compile().unwrap(); assert_eq!( plan.precompute_plan.materializations.len(), 1, @@ -131,7 +131,8 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write() ); let output = tempfile::tempdir().unwrap(); let path = output.path().join("planning.json"); - std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap(); + let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap()); + std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap(); let port = unused_port(); let mut child = ChildGuard( Command::new(env!("CARGO_BIN_EXE_data_plane")) diff --git a/data_plane/tests/support/immutable_maintenance_process.rs b/data_plane/tests/support/immutable_maintenance_process.rs index eb191c0a..f105cdfa 100644 --- a/data_plane/tests/support/immutable_maintenance_process.rs +++ b/data_plane/tests/support/immutable_maintenance_process.rs @@ -35,7 +35,7 @@ async fn run_maintenance_process(multi_source: bool, distinct_groups: bool) { fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(fixture.clone()).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = quote_snapshot_for_test(snapshot).compile().unwrap(); assert_eq!( plan.precompute_plan.materializations.len(), if multi_source { 3 } else { 2 } @@ -303,12 +303,15 @@ async fn run_maintenance_process(multi_source: bool, distinct_groups: bool) { // singleton-derived output while new source populations can arrive. let mut next_fixture = fixture.clone(); next_fixture["environment"]["plan_version"] = 2.into(); - let next = serde_json::from_value::< - control_plane::physical::compiler::BackendLocalPlanningSnapshot, - >(next_fixture) - .unwrap() - .compile() - .unwrap(); + let next = + quote_snapshot_for_test( + serde_json::from_value::< + control_plane::physical::compiler::BackendLocalPlanningSnapshot, + >(next_fixture) + .unwrap(), + ) + .compile() + .unwrap(); let next_install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { summary_catalog: next.summary_catalog, collector_plans: next.collector_plans, diff --git a/data_plane/tests/support/univmon_erp_process.rs b/data_plane/tests/support/univmon_erp_process.rs index 9c933b8c..7c496d4a 100644 --- a/data_plane/tests/support/univmon_erp_process.rs +++ b/data_plane/tests/support/univmon_erp_process.rs @@ -125,7 +125,7 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { "runtime": {"allowed_algorithms": ["Hll", "Kll", "UnivMon"], "max_memory_bytes": null} }); let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture.clone()).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = quote_snapshot_for_test(snapshot).compile().unwrap(); eprintln!( "UNIVMON_PLANNED {}", serde_json::json!({"query_plan": plan.query_plan, "materializations": plan.precompute_plan.materializations, "lifecycle_estimates": plan.lifecycle_estimates, "executable_dags": plan.precompute_plan.executable_dags, "observation": observation}) @@ -148,10 +148,11 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { .unwrap() .remove("max_frequency_entropy_absolute_bits_error"); } - let missing = serde_json::from_value::(missing_entropy) - .unwrap() - .compile() - .unwrap(); + let missing = quote_snapshot_for_test( + serde_json::from_value::(missing_entropy).unwrap(), + ) + .compile() + .unwrap(); use control_plane::query_plan::{QueryPlanNode, QueryReadout}; assert!(missing .query_plan @@ -210,7 +211,8 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { }); let output = tempfile::tempdir().unwrap(); let path = output.path().join("planning.json"); - std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap(); + let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap()); + std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap(); let port = unused_port(); let mut child = ChildGuard( Command::new(env!("CARGO_BIN_EXE_data_plane")) @@ -326,7 +328,7 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { .unwrap() .invalid_reason .is_none()); - let replanned = live_snapshot.compile().unwrap(); + let replanned = quote_snapshot_for_test(live_snapshot).compile().unwrap(); assert!( replanned .precompute_plan diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json index e345114c..8641f93b 100644 --- a/docs/examples/asapquery-compatibility-demo-snapshot.json +++ b/docs/examples/asapquery-compatibility-demo-snapshot.json @@ -1,5 +1,5 @@ { - "snapshot_version": 1, + "snapshot_version": 2, "query_workload": { "language": "promql", "query_batch": null, diff --git a/docs/examples/asapquery-planning-snapshot.json b/docs/examples/asapquery-planning-snapshot.json index 99c2a284..cd63e8bc 100644 --- a/docs/examples/asapquery-planning-snapshot.json +++ b/docs/examples/asapquery-planning-snapshot.json @@ -1,5 +1,5 @@ { - "snapshot_version": 1, + "snapshot_version": 2, "query_workload": { "language": "promql", "query_batch": null, From cff91375d330f78f4d11d79e3354e1d7fb3f4102 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 11:22:18 -0600 Subject: [PATCH 2/2] Unify snapshot deployment CLI and migrate discovery examples --- README.md | 31 ++++++------ .../docs/candidate-physical-explain.md | 2 +- .../examples/compile_workload_artifact.rs | 23 ++++++--- .../examples/inspect_physical_dag.rs | 49 ------------------- demos/asapquery/run.sh | 3 +- docs/evaluation/e2e-physical-dag.md | 26 +++++----- docs/examples/workload-cost-evidence.md | 8 +-- 7 files changed, 53 insertions(+), 89 deletions(-) delete mode 100644 control_plane/examples/inspect_physical_dag.rs diff --git a/README.md b/README.md index 57181620..102e1783 100644 --- a/README.md +++ b/README.md @@ -264,7 +264,7 @@ docker version cargo +1.98.0 fetch --locked cargo +1.98.0 build --locked -p control_plane -p data_plane cargo +1.98.0 build --locked -p control_plane \ - --example calibration_candidates --example inspect_physical_dag \ + --example calibration_candidates --example compile_workload_artifact \ --example compile_clickhouse_workload target/debug/data_plane --help mkdir -p target/readme-evidence @@ -278,7 +278,7 @@ mkdir -p target/readme-evidence The executable is `target/debug/data_plane`, or `target/release/data_plane` if you build with `--release`. Access to the pinned Git dependencies is required. -### 3. Export candidates and inspect the ordinary selected plan +### 3. Export candidates and select a deployable plan by complete cost ```bash target/debug/examples/calibration_candidates \ @@ -287,10 +287,10 @@ target/debug/examples/calibration_candidates \ jq '.candidates[] | {candidate_index, unavailable_reason}' \ target/readme-evidence/candidates.json -target/debug/examples/inspect_physical_dag \ - docs/examples/asapquery-compatibility-demo-snapshot.json \ +target/debug/examples/compile_workload_artifact \ + "$ASAPQUERY_PLANNING_SNAPSHOT" \ > target/readme-evidence/selected.json -jq '.purpose' target/readme-evidence/selected.json +jq '.cost_comparison' target/readme-evidence/selected.json jq '.install_request.summary_catalog' target/readme-evidence/selected.json jq '.install_request.precompute_plan | {materializations, executable_dags}' \ target/readme-evidence/selected.json @@ -299,13 +299,13 @@ jq '.install_request.precompute_plan.schemas[] | {materialization, schema_id}' \ target/readme-evidence/selected.json ``` -Expect `inspection_only`, catalog/plan objects and explicit candidate rejection -reasons where unsupported. One verified demo export contained five candidate -entries (one installable), five selected materializations and six query entries; -these are inspection evidence, not a permanent optimizer-count contract. -Materialization IDs are definitions, not physical SIDs. Demo costs are not -measurements. The [E2E walkthrough](docs/evaluation/e2e-physical-dag.md) explains -ERP evidence and the version-2 measured-cost workflow. +Candidate discovery accepts the checked-in unquoted templates. Deployment and +selected-plan inspection require `ASAPQUERY_PLANNING_SNAPSHOT` to point to a +snapshot with complete, valid workload cost evidence. Prepare that input using +the [cost evidence workflow](docs/examples/workload-cost-evidence.md). +There is one snapshot compiler: it compares complete executable alternatives, +including exact fallback. Materialization IDs are definitions, not physical SIDs. +For `--metricsql`, collect quotes for the MetricsQL frontend. ## Prometheus runbook @@ -342,7 +342,7 @@ done curl -fsS http://127.0.0.1:19090/-/healthy target/debug/data_plane --profile asapquery \ - --planning-snapshot docs/examples/asapquery-compatibility-demo-snapshot.json \ + --planning-snapshot "$ASAPQUERY_PLANNING_SNAPSHOT" \ --prometheus-server http://127.0.0.1:19090 \ --forward-unsupported-queries --http-port 19091 \ --output-dir target/readme-evidence/prometheus/runtime \ @@ -409,6 +409,7 @@ kill "$(cat target/readme-evidence/prometheus/backend.pid)" docker rm -f asap-readme-prometheus asap-readme-pushgateway cargo +1.98.0 test --locked -p data_plane --test asapquery_compatibility_process_e2e \ collector_free_profile_serves_complete_matrix_and_falls_back_exactly -- --exact +export ASAPQUERY_PLANNING_SNAPSHOT=/absolute/path/priced-snapshot.json ./scripts/e2e.sh asapquery-demo ``` @@ -467,8 +468,8 @@ kill "$(cat target/readme-evidence/victoriametrics/backend.pid)" To inspect supported MetricsQL planning independently: ```bash -target/debug/examples/inspect_physical_dag \ - docs/examples/asapquery-compatibility-demo-snapshot.json --metricsql \ +target/debug/examples/compile_workload_artifact \ + "$ASAPQUERY_PLANNING_SNAPSHOT" --metricsql \ > target/readme-evidence/victoriametrics/selected.json jq '.install_request.query_plan.entries' \ target/readme-evidence/victoriametrics/selected.json diff --git a/control_plane/docs/candidate-physical-explain.md b/control_plane/docs/candidate-physical-explain.md index 993ce39f..fded1a0a 100644 --- a/control_plane/docs/candidate-physical-explain.md +++ b/control_plane/docs/candidate-physical-explain.md @@ -18,7 +18,7 @@ Physical identity combines the logical/mask alternative with existing materializ The read-only `/api/v1/physical-plan/cost-manifests` and MetricsQL equivalent retain their default manifest-array response. Add `"explain": true` to the existing request to receive `{ "manifests": [...], "alternatives": [...], "logical_selection": [...] }`. Failed alternatives remain alongside usable manifests. When none can bind or be completely priced, the error retains an `all_infeasible` report and every accumulated alternative rather than only a generic message. -Snapshot compilation exposes the same logical trace on `PhysicalPlan`; `inspect_physical_dag` prints it outside the install request. Compile-and-publish returns the trace without adding it to executable wire DTOs. SQL's existing selection trace gains the same semantic candidate/root identities. +Snapshot compilation exposes the same logical trace on `PhysicalPlan`; `compile_workload_artifact` prints it outside the install request. Compile-and-publish returns the trace without adding it to executable wire DTOs. SQL's existing selection trace gains the same semantic candidate/root identities. These are bounded explanations: they cover the actual Planner search and the existing physical materialization/exact inventory, not every possible placement or resource-constrained cluster assignment. Missing numeric measurements remain missing. The next provider integration must occur before logical commitment and reuse Planner's provider/resource contracts. diff --git a/control_plane/examples/compile_workload_artifact.rs b/control_plane/examples/compile_workload_artifact.rs index b9c3c348..fa6ceed5 100644 --- a/control_plane/examples/compile_workload_artifact.rs +++ b/control_plane/examples/compile_workload_artifact.rs @@ -3,17 +3,21 @@ use control_plane::physical::compiler::BackendLocalPlanningSnapshot; use serde_json::json; fn main() -> Result<(), Box> { - let path = std::env::args() - .nth(1) + let mut args = std::env::args().skip(1); + let path = args + .next() .ok_or("usage: compile_workload_artifact SNAPSHOT.json [--metricsql]")?; - let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?; - if snapshot.snapshot_version != 2 { - return Err( - "execution evaluation requires version 2 complete workload cost evidence".into(), - ); + let metricsql = match args.next().as_deref() { + None => false, + Some("--metricsql") => true, + Some(_) => return Err("expected optional --metricsql".into()), + }; + if args.next().is_some() { + return Err("unexpected arguments".into()); } + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?; let start = std::time::Instant::now(); - let plan = if std::env::args().skip(2).any(|arg| arg == "--metricsql") { + let plan = if metricsql { snapshot.compile_metricsql()? } else { snapshot.compile()? @@ -29,6 +33,9 @@ fn main() -> Result<(), Box> { "planning_elapsed_ns": elapsed, "envelope": plan.envelope, "cost_comparison": comparison, + "logical_selection": plan.logical_selection, + "backend_revision": control_plane::physical::compiler::BACKEND_REVISION, + "planner_revision": control_plane::physical::compiler::PLANNER_REVISION, "lifecycle_estimates": plan.lifecycle_estimates, "install_request": { "summary_catalog": plan.summary_catalog, diff --git a/control_plane/examples/inspect_physical_dag.rs b/control_plane/examples/inspect_physical_dag.rs deleted file mode 100644 index b9d812ff..00000000 --- a/control_plane/examples/inspect_physical_dag.rs +++ /dev/null @@ -1,49 +0,0 @@ -//! Inspect an ordinary planning snapshot without treating demo costs as measurements. -use control_plane::physical::compiler::BackendLocalPlanningSnapshot; -use serde_json::json; - -fn main() -> Result<(), Box> { - let mut args = std::env::args().skip(1); - let path = args - .next() - .ok_or("usage: inspect_physical_dag SNAPSHOT.json [--metricsql]")?; - let metricsql = match args.next().as_deref() { - None => false, - Some("--metricsql") => true, - Some(_) => return Err("expected optional --metricsql".into()), - }; - if args.next().is_some() { - return Err("unexpected arguments".into()); - } - let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?; - let snapshot_version = snapshot.snapshot_version; - let erp_supplied = snapshot.implementation.erp.is_some(); - let plan = if metricsql { - snapshot.compile_metricsql()? - } else { - snapshot.compile()? - }; - println!( - "{}", - serde_json::to_string_pretty(&json!({ - "purpose": "inspection_only", - "snapshot_version": snapshot_version, - "erp_input_supplied": erp_supplied, - "backend_revision": control_plane::physical::compiler::BACKEND_REVISION, - "planner_revision": control_plane::physical::compiler::PLANNER_REVISION, - "cost_comparison": plan.cost_comparison, - "logical_selection": plan.logical_selection, - "lifecycle_estimates": plan.lifecycle_estimates, - "install_request": { - "summary_catalog": plan.summary_catalog, - "collector_plans": plan.collector_plans, - "precompute_plan": plan.precompute_plan, - "transmission_plan": plan.transmission_plan, - "query_plan": plan.query_plan, - "storage_routing": null, - "adaptation_evidence": [] - } - }))? - ); - Ok(()) -} diff --git a/demos/asapquery/run.sh b/demos/asapquery/run.sh index addad6b5..95546995 100755 --- a/demos/asapquery/run.sh +++ b/demos/asapquery/run.sh @@ -11,6 +11,7 @@ PROM_CONTAINER="${RUN_ID}-prometheus" PUSH_CONTAINER="${RUN_ID}-pushgateway" EVIDENCE_DIR="${ASAPQUERY_DEMO_EVIDENCE_DIR:-${REPO_DIR}/target/asapquery-demo-evidence}" BACKEND_PID="" +: "${ASAPQUERY_PLANNING_SNAPSHOT:?Set ASAPQUERY_PLANNING_SNAPSHOT to a snapshot with complete workload cost evidence}" cleanup() { if [[ -n "${BACKEND_PID}" ]]; then @@ -52,7 +53,7 @@ curl -fsS http://127.0.0.1:19090/-/healthy >/dev/null "${REPO_DIR}/target/debug/data_plane" \ --profile asapquery \ - --planning-snapshot "${REPO_DIR}/docs/examples/asapquery-compatibility-demo-snapshot.json" \ + --planning-snapshot "${ASAPQUERY_PLANNING_SNAPSHOT}" \ --prometheus-server http://127.0.0.1:19090 \ --forward-unsupported-queries \ --http-port 19091 \ diff --git a/docs/evaluation/e2e-physical-dag.md b/docs/evaluation/e2e-physical-dag.md index c17d609f..1a3c2807 100644 --- a/docs/evaluation/e2e-physical-dag.md +++ b/docs/evaluation/e2e-physical-dag.md @@ -39,10 +39,9 @@ Planner owns query semantics, summary families, parameters and candidate selection. ERP can affect supported evidence-based choices, but supplying an artifact does not prove that it was eligible or used. Freshness, source/update semantics, parameters and accuracy constraints still apply. The checked-in demo -is not an empirical-ERP benchmark. For measured cost selection use the version-2 -snapshot workflow in [execution calibration](../../tools/o11y-execution/CALIBRATION.md). -`compile_workload_artifact` deliberately requires that version; do not bypass it -by relabeling demo costs as measured evidence. +is not an empirical-ERP benchmark. For deployment, use the priced snapshot workflow in [execution calibration](../../tools/o11y-execution/CALIBRATION.md). +`compile_workload_artifact` requires complete workload quotes; do not relabel +demo costs as measured evidence. For the existing observation → ERP-selected KLL → installed HTTP correctness fixture, see [ERP process validation](../developer_docs/erp-process-validation.md): @@ -62,8 +61,8 @@ Run the normal snapshot compiler, without selecting a candidate index or substituting a preferred sketch family: ```bash -cargo run --locked -p control_plane --example inspect_physical_dag -- \ - docs/examples/asapquery-compatibility-demo-snapshot.json \ +cargo run --locked -p control_plane --example compile_workload_artifact -- \ + "$ASAPQUERY_PLANNING_SNAPSHOT" \ > target/physical-dag-inspection/selected.json jq '.install_request' target/physical-dag-inspection/selected.json \ > target/physical-dag-inspection/physical-plan.json @@ -71,10 +70,12 @@ jq '.install_request | {summary_catalog, precompute_plan, query_plan}' \ target/physical-dag-inspection/selected.json ``` -`inspect_physical_dag` calls the same `BackendLocalPlanningSnapshot::compile` -entry point used by startup. It accepts the checked-in demonstration snapshot -and labels its output `inspection_only`. `erp_input_supplied` reports only input -presence. A null `cost_comparison` must not be interpreted as a measured win. +`compile_workload_artifact` calls the same evidence-required snapshot compiler +used by startup. Set `ASAPQUERY_PLANNING_SNAPSHOT` to a priced snapshot prepared +using the [cost evidence workflow](../examples/workload-cost-evidence.md). +The checked-in unquoted templates support candidate discovery only. The output +includes the selected plan, logical selection trace and complete cost comparison. +MetricsQL compilation requires quotes collected for that frontend. The compiler derives sibling plans from the selected post-ASAP DAG: @@ -95,6 +96,7 @@ participate in identity, cost and coverage checks. For the simplest live run: ```bash +export ASAPQUERY_PLANNING_SNAPSHOT=/absolute/path/priced-snapshot.json ./scripts/e2e.sh asapquery-demo ``` @@ -116,7 +118,7 @@ Then keep the backend running for inspection: ```bash cargo run --locked -p data_plane -- \ --profile asapquery \ - --planning-snapshot docs/examples/asapquery-compatibility-demo-snapshot.json \ + --planning-snapshot "$ASAPQUERY_PLANNING_SNAPSHOT" \ --prometheus-server http://127.0.0.1:9090 \ --forward-unsupported-queries --http-port 9091 \ --output-dir target/physical-dag-inspection/runtime @@ -242,7 +244,7 @@ when comparing, `comparison.json`. A finished run is not proof of benefit. | Backend | Drop-in surface and current boundary | | --- | --- | | Prometheus | Remote Write v1 samples plus PromQL instant/range HTTP; keep Prometheus as the exact fallback and raw-data authority. The strict `asapquery` profile has explicit startup exclusions. | -| VictoriaMetrics | The broader backend has a MetricsQL compile/adapter path. `inspect_physical_dag SNAPSHOT.json --metricsql` uses it. This does not turn the strict Prometheus profile into a general VM replacement; counter boundary semantics and unsupported expressions must retain exact routing. | +| VictoriaMetrics | The broader backend has a MetricsQL compile/adapter path. `compile_workload_artifact SNAPSHOT.json --metricsql` uses it. This does not turn the strict Prometheus profile into a general VM replacement; counter boundary semantics and unsupported expressions must retain exact routing. | | ClickHouse | The broader backend has a SQL workload compiler and typed exact/relational execution. `compile_clickhouse_workload` reads its own `ClickHouseSqlAutomaticWorkload` JSON from stdin; it does not consume the Prometheus snapshot. List/Map/Tuple support does not imply arbitrary lambdas/counter SQL or full protocol compatibility. External-only DAG coverage is not summary acceleration. | Use the existing process suites to validate a specific supported protocol shape; diff --git a/docs/examples/workload-cost-evidence.md b/docs/examples/workload-cost-evidence.md index 612f58e7..64e60498 100644 --- a/docs/examples/workload-cost-evidence.md +++ b/docs/examples/workload-cost-evidence.md @@ -1,8 +1,10 @@ # 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. +Planning snapshots use one schema, `snapshot_version: 2`. Version 1 is rejected. +Candidate discovery may omit `workload_cost_evidence`; compiling a deployable +snapshot requires complete, valid quotes and selects by complete workload cost. +There is no unquoted snapshot deployment path. The checked-in JSON examples are +discovery templates, not ready-to-deploy plans. ## Workflow