diff --git a/Cargo.lock b/Cargo.lock index a49ca3f3..4ba3a108 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -364,7 +364,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cd7e9e0f710816d49190dabd6c789359067a208f#cd7e9e0f710816d49190dabd6c789359067a208f" dependencies = [ "asap-types", "asap_sketchlib 0.3.0 (git+https://github.com/ProjectASAP/asap_sketchlib)", @@ -376,7 +376,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cd7e9e0f710816d49190dabd6c789359067a208f#cd7e9e0f710816d49190dabd6c789359067a208f" dependencies = [ "asap-types", "promql-parser 0.10.0 (git+https://github.com/ProjectASAP/promql-parser?rev=9fede7eecca923c9882fe256484d00d37f8706cb)", @@ -385,7 +385,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cd7e9e0f710816d49190dabd6c789359067a208f#cd7e9e0f710816d49190dabd6c789359067a208f" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -408,12 +408,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cd7e9e0f710816d49190dabd6c789359067a208f#cd7e9e0f710816d49190dabd6c789359067a208f" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cd7e9e0f710816d49190dabd6c789359067a208f#cd7e9e0f710816d49190dabd6c789359067a208f" dependencies = [ "serde", "serde_json", @@ -1660,7 +1660,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -2277,7 +2277,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -2304,15 +2304,6 @@ dependencies = [ "either", ] -[[package]] -name = "itertools" -version = "0.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" -dependencies = [ - "either", -] - [[package]] name = "itoa" version = "1.0.18" @@ -3084,7 +3075,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck", - "itertools 0.14.0", + "itertools 0.10.5", "log", "multimap", "once_cell", @@ -3104,7 +3095,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.10.5", "proc-macro2", "quote", "syn 2.0.119", @@ -3242,7 +3233,7 @@ dependencies = [ "once_cell", "socket2 0.5.10", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -3518,7 +3509,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -3963,7 +3954,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -4475,7 +4466,7 @@ version = "2.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5283634e518fe9e82c7b20520bb4bc209009fd16c82077c802f8111ecbb0117a" dependencies = [ - "rand 0.9.5", + "rand 0.10.2", ] [[package]] @@ -4738,7 +4729,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index f1da29da..eaca3523 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,10 +20,10 @@ asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", branch [workspace.dependencies] # Keep Planner frontends, selection, and IR on the same immutable revision. # Alias upstream asap-types because this workspace also defines asap_types. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "10d93846f0ef9f06ab19e599701d699d9cdd15af" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "10d93846f0ef9f06ab19e599701d699d9cdd15af" } -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "10d93846f0ef9f06ab19e599701d699d9cdd15af" } -asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "10d93846f0ef9f06ab19e599701d699d9cdd15af" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cd7e9e0f710816d49190dabd6c789359067a208f" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cd7e9e0f710816d49190dabd6c789359067a208f" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cd7e9e0f710816d49190dabd6c789359067a208f" } +asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cd7e9e0f710816d49190dabd6c789359067a208f" } # Shared external deps (used by 2+ crates) serde = { version = "1.0", features = ["derive"] } diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index ef76bf73..b0b3bf86 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -645,11 +645,12 @@ fn compile_physical_plan_request( ) .map_err(|error| (StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into()))?; let planner_selection_trace = compilation_request.planner_selection_trace.clone(); - let (manifests, alternatives) = physical::workload_cost::compile_candidates_for_pricing( - candidates.clone(), - environment.clone(), - frontend, - ); + let (manifests, candidate_evaluations) = + physical::workload_cost::compile_candidates_for_pricing( + candidates.clone(), + environment.clone(), + frontend, + ); 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. @@ -657,7 +658,7 @@ fn compile_physical_plan_request( if manifests.is_empty() { return Err(( StatusCode::UNPROCESSABLE_ENTITY, - serde_json::json!({"status": "all_infeasible", "alternatives": alternatives, + serde_json::json!({"status": "all_infeasible", "candidates": candidate_evaluations, "logical_selection": compilation_request.planner_selection_trace}), )); } @@ -666,7 +667,7 @@ fn compile_physical_plan_request( request.target_collector_ids, apply_timeout, request.runtime_adaptation_evidence, - (manifests, alternatives, planner_selection_trace), + (manifests, candidate_evaluations, planner_selection_trace), )); } let compiled = match request.workload_cost_evidence { @@ -688,7 +689,7 @@ fn compile_physical_plan_request( }; let bundle = match compiled { Ok(bundle) => bundle, - Err(physical::compiler::CompileError::Alternatives(report)) => { + Err(physical::compiler::CompileError::Candidates(report)) => { return Err((StatusCode::UNPROCESSABLE_ENTITY, report)) } Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into())), @@ -698,7 +699,7 @@ fn compile_physical_plan_request( request.target_collector_ids, apply_timeout, request.runtime_adaptation_evidence, - (manifests, alternatives, planner_selection_trace), + (manifests, candidate_evaluations, planner_selection_trace), )) } @@ -735,9 +736,9 @@ fn workload_cost_manifests( } let explain = request.explain; match compile_physical_plan_request(request, true, frontend) { - Ok((_, _, _, _, (manifests, alternatives, planner_selection_trace))) => { + Ok((_, _, _, _, (manifests, candidates, planner_selection_trace))) => { if explain { - Json(serde_json::json!({"manifests": manifests, "alternatives": alternatives, "logical_selection": planner_selection_trace})) + Json(serde_json::json!({"manifests": manifests, "candidates": candidates, "logical_selection": planner_selection_trace})) .into_response() } else { Json(manifests).into_response() @@ -866,12 +867,12 @@ mod api_tests { let manifests = body_json(response).await; if explain { assert_eq!(manifests["manifests"].as_array().unwrap().len(), 1); - let alternatives = manifests["alternatives"].as_array().unwrap(); - assert_eq!(alternatives.len(), 2); - assert_eq!(alternatives[0]["status"], "bind_failed"); - assert!(alternatives[0]["unavailable_reason"].is_string()); - assert_eq!(alternatives[1]["status"], "bound"); - assert!(alternatives[1]["physical_alternative_id"].is_string()); + let candidates = manifests["candidates"].as_array().unwrap(); + assert_eq!(candidates.len(), 2); + assert_eq!(candidates[0]["status"], "bind_failed"); + assert!(candidates[0]["unavailable_reason"].is_string()); + assert_eq!(candidates[1]["status"], "bound"); + assert!(candidates[1]["physical_candidate_id"].is_string()); assert!(!manifests["logical_selection"] .as_array() .unwrap() @@ -889,8 +890,8 @@ mod api_tests { assert_eq!(response.headers()["content-type"], "application/json"); let report = body_json(response).await; assert_eq!(report["status"], "all_infeasible"); - assert_eq!(report["alternatives"].as_array().unwrap().len(), 2); - assert!(report["alternatives"] + assert_eq!(report["candidates"].as_array().unwrap().len(), 2); + assert!(report["candidates"] .as_array() .unwrap() .iter() diff --git a/control_plane/src/physical/backend_stage.rs b/control_plane/src/physical/backend_stage.rs index a9a64f6a..073ea1d6 100644 --- a/control_plane/src/physical/backend_stage.rs +++ b/control_plane/src/physical/backend_stage.rs @@ -1,7 +1,7 @@ //! Backend-facing projection of one planning cycle. //! //! These types are the input to [`crate::backend_plan::from_stage_config`] and -//! to `emit::backend_wire`'s backend JSON builders. `PhysicalCompiler` builds +//! to `emit::backend_wire`'s backend JSON builders. `PhysicalPlanCompiler` builds //! them directly from the summaries ASAPPlanner selected. //! //! They are deliberately not `Serialize`/`Deserialize`: `SummaryFamilyType` diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 747fc171..69254a73 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1,6 +1,6 @@ //! Backend-owned physical compilation over ASAPPlanner's selected post-ASAP IR. //! -//! Planner owns semantic alternatives and guarantees. This module owns the +//! Planner owns semantic candidates and guarantees. This module owns the //! deployment decision: evidence freshness, target capabilities, windows, the //! Collector execution projection, SummaryCatalog, and executable plans. @@ -8,13 +8,13 @@ use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::rc::Rc; use asap_aware_mapping::cost_model::Cost; +#[cfg(test)] +use asap_aware_mapping::DefaultAccuracyModel; use asap_aware_mapping::{ plan_summary_maintenance_lifecycles, AccuracyEvidenceProvider, CostRate, Horizon, PropagationStats, SummaryMaintenanceCapabilities, SummaryMaintenanceLifecycleCapabilities, SummaryMaintenanceLifecycleCostInputs, WorkloadDemand, }; -#[cfg(test)] -use asap_aware_mapping::{DefaultAccuracyModel, EqualSplitAllocator}; use planner_types::post_asap::{ CompositionOperator, EvaluationSchedule, ExecutableDagCompilation, OutputRepresentation, PostAsapNodeId, SketchAlgorithm, SketchParams, SketchQuery, SummaryExpr, SummaryFamilyType, @@ -114,7 +114,7 @@ pub struct WindowRealizationCostQuote { #[serde(deny_unknown_fields)] pub struct WindowRealizationCandidate { /// Backend-owned identity; never copied into Planner IR. - #[serde(rename = "implementation_id", alias = "realization_id")] + #[serde(rename = "implementation_id")] pub realization_id: String, pub framework: SummaryWindowFramework, pub window_secs: u64, @@ -200,7 +200,7 @@ pub struct TopKMembershipEvidence { #[serde(deny_unknown_fields)] pub struct PhysicalDeploymentContext { pub target: PhysicalDeploymentTarget, - #[serde(rename = "collector_ids", alias = "target_collector_ids")] + #[serde(rename = "collector_ids")] pub target_collector_ids: Vec, pub capability_snapshot_id: String, pub observed_at_unix_ms: u64, @@ -226,7 +226,7 @@ pub enum PhysicalDeploymentTarget { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] pub struct BackendLocalPlanningInput { - #[serde(rename = "snapshot_version", alias = "schema_version")] + #[serde(rename = "snapshot_version")] pub schema_version: u32, /// May be absent during candidate discovery, never during deployment. #[serde(default, skip_serializing_if = "Option::is_none")] @@ -234,7 +234,7 @@ pub struct BackendLocalPlanningInput { #[serde(deserialize_with = "deserialize_snapshot_query_workload")] pub query_workload: QueryWorkload, pub data_workload: DataWorkload, - #[serde(rename = "implementation", alias = "physical_inputs")] + #[serde(rename = "implementation")] pub physical_inputs: BackendLocalPhysicalInputs, pub environment: PhysicalDeploymentContext, } @@ -252,22 +252,15 @@ pub struct BackendLocalPhysicalInputs { /// default is distinct from Prometheus instant-selector lookback delta. pub scrape_interval_ms: u64, #[serde(default, skip_serializing_if = "u64_is_zero")] - #[serde( - rename = "query_staleness_margin_ms", - alias = "query_retention_margin_ms" - )] + #[serde(rename = "query_staleness_margin_ms")] pub query_retention_margin_ms: u64, /// Admission budget for all retained panes and estimated partitions. /// Missing legacy snapshots inherit the backend default. #[serde( default = "default_retained_summary_memory_budget_bytes", - alias = "maxRetainedSummaryBytes", skip_serializing_if = "is_default_retained_summary_memory_budget_bytes" )] - #[serde( - rename = "max_retained_summary_bytes", - alias = "retained_summary_memory_budget_bytes" - )] + #[serde(rename = "max_retained_summary_bytes")] pub retained_summary_memory_budget_bytes: u64, /// Certificates keyed by exact registered PromQL; converted to root IDs /// before workload selection so one query cannot borrow another's evidence. @@ -481,7 +474,7 @@ pub struct CompiledPhysicalPlan { pub struct MaterializationLifecycleEstimate { pub materialization: asap_types::sds::SummaryDefinitionId, pub consumer_query_ids: Vec, - #[serde(rename = "window_implementation_id", alias = "window_realization_id")] + #[serde(rename = "window_implementation_id")] pub window_realization_id: String, pub horizon_seconds: f64, pub expected_reads: f64, @@ -494,7 +487,7 @@ pub enum CompileError { #[error("invalid backend-local workload snapshot: {0}")] Snapshot(String), #[error("no feasible completely costed alternative: {0}")] - Alternatives(serde_json::Value), + Candidates(serde_json::Value), #[error("planner revision mismatch: request={request}, compiler={compiler}")] PlannerRevision { request: String, @@ -1090,6 +1083,7 @@ impl PhysicalPlanCompiler { validate_evidence(&query.query_id, e, &environment)?; } let node = query.selected_plan_root.clone(); + reject_uncertified_readouts(&query.query_id, &node)?; let selected = collect_selected_materializations( &node, request.allow_mixed_summary_and_exact_execution, @@ -2383,11 +2377,10 @@ pub fn select_post_asap( delete: false, }, ); - crate::planner_selection::select_summary_with_evidence( + crate::planner_selection::select_query_with_models( expr, &model, &DefaultAccuracyModel, - &EqualSplitAllocator, &QueryEvidence(evidence), ) } @@ -2793,20 +2786,25 @@ pub(super) fn retained_state_count( /// cells use two words here, covering the counter plus observed serialization /// overhead. Heap and exact-state estimates include container slack. fn retained_state_bytes(materialization: &asap_types::PrecomputeMaterialization) -> u128 { + estimated_state_bytes( + &materialization.aggregation_type, + &materialization.parameters, + ) +} + +pub(super) fn estimated_state_bytes( + aggregation: &asap_types::AggregationType, + parameters: &HashMap, +) -> u128 { use asap_types::AggregationType as A; let parameter = |names: &[&str], fallback: u64| { names .iter() - .find_map(|name| { - materialization - .parameters - .get(*name) - .and_then(Value::as_u64) - }) + .find_map(|name| parameters.get(*name).and_then(Value::as_u64)) .unwrap_or(fallback) as u128 }; - match materialization.aggregation_type { + match aggregation { A::CountMinSketch | A::CountSketch => { parameter(&["width", "w", "col_num", "col"], 1) * parameter(&["depth", "d", "row_num", "row"], 1) @@ -3261,6 +3259,31 @@ fn validate_executable_subdag(node: &Rc) -> Result<(), String> { Ok(()) } +fn reject_uncertified_readouts(query_id: &str, root: &Rc) -> Result<(), CompileError> { + let dag = planner_types::post_asap::compile_executable_dag(root).map_err(|error| { + CompileError::Query { + query_id: query_id.into(), + reason: format!("invalid executable subDAG: {error}"), + } + })?; + for node in &dag.nodes { + if matches!( + node.payload, + planner_types::post_asap::ExecutableOperatorPayload::SummaryEstimate { .. } + ) && node + .guarantee + .as_ref() + .is_none_or(planner_types::post_asap::ResultGuarantee::has_unknown) + { + return Err(CompileError::Query { + query_id: query_id.into(), + reason: "selected summary readout has no certified accuracy guarantee; provide scoped evidence or use exact execution".into(), + }); + } + } + Ok(()) +} + fn physical_aggregation( query: &QueryCompilationInput, selected: &SelectedMaterialization, @@ -3700,7 +3723,7 @@ pub(crate) fn physical_materialization_family(family: &SummaryFamilyType) -> Sum } } -fn sketch_params_json(params: &planner_types::post_asap::SketchParams) -> Value { +pub(super) fn sketch_params_json(params: &planner_types::post_asap::SketchParams) -> Value { use planner_types::post_asap::SketchParams as P; match params { P::UnivMon { @@ -3751,30 +3774,6 @@ fn stable_workload_plan_id( hasher.finish() } -// Compatibility imports; new callers use the domain names above. -#[deprecated(note = "Use BackendLocalPhysicalInputs")] -pub use BackendLocalPhysicalInputs as BackendLocalImplementation; -#[deprecated(note = "Use BackendLocalPlanningInput")] -pub use BackendLocalPlanningInput as BackendLocalPlanningSnapshot; -#[deprecated(note = "Use CompiledPhysicalPlan")] -pub use CompiledPhysicalPlan as PhysicalPlan; -#[deprecated(note = "Use LifecycleUnitCosts")] -pub use LifecycleUnitCosts as LifecycleCostEvidence; -#[deprecated(note = "Use PhysicalCompilationRequest")] -pub use PhysicalCompilationRequest as PlanningRequest; -#[deprecated(note = "Use PhysicalDeploymentContext")] -pub use PhysicalDeploymentContext as DeploymentEnvironment; -#[deprecated(note = "Use PhysicalPlanCompiler")] -pub use PhysicalPlanCompiler as PhysicalCompiler; -#[deprecated(note = "Use QueryCompilationInput")] -pub use QueryCompilationInput as PlanningQuery; -#[deprecated(note = "Use SummaryLifecyclePlanningInputs")] -pub use SummaryLifecyclePlanningInputs as LifecyclePlanningInput; -#[deprecated(note = "Use WindowRealizationCandidate")] -pub use WindowRealizationCandidate as WindowImplementationCandidate; -#[deprecated(note = "Use WindowRealizationCostQuote")] -pub use WindowRealizationCostQuote as ImplementationCostEvidence; - /// Parser/executor frontend for the time-series physical compiler. SQL has its /// own compilation input and must not silently enter this path. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -3802,38 +3801,6 @@ impl QueryFrontend { PhysicalPlanCompiler.compile_for_frontend(request, environment, self) } } -impl BackendLocalPlanningInput { - #[deprecated(note = "Use compile_promql")] - pub fn compile(self) -> Result { - self.compile_promql() - } - #[deprecated(note = "Use into_physical_compilation_request")] - pub fn planning_request( - self, - ) -> Result<(PhysicalCompilationRequest, PhysicalDeploymentContext), CompileError> { - self.into_physical_compilation_request() - } -} -impl PhysicalPlanCompiler { - #[deprecated(note = "Use compile_promql")] - pub fn compile( - &self, - request: PhysicalCompilationRequest, - environment: PhysicalDeploymentContext, - ) -> Result { - self.compile_promql(request, environment) - } -} - -#[deprecated(note = "Use build_transmission_plan")] -pub use build_transmission_plan as compile_transmission_plan; -#[deprecated(note = "Use select_logical_roots_for_queries")] -pub use select_logical_roots_for_queries as select_workload_roots; -#[deprecated(note = "Use select_logical_roots_with_error_resource_profiles")] -pub use select_logical_roots_with_error_resource_profiles as select_workload_roots_with_erp; -#[deprecated(note = "Use select_logical_roots_with_trace")] -pub use select_logical_roots_with_trace as select_workload_roots_with_trace; - #[cfg(test)] pub(crate) mod tests { use super::*; @@ -3867,15 +3834,16 @@ pub(crate) mod tests { .collect(), ); let (request, environment) = snapshot.into_physical_compilation_request().unwrap(); - let plans: Vec<_> = super::super::workload_cost::with_exact_alternative(request) - .unwrap() - .into_iter() - .filter_map(|r| { - PhysicalPlanCompiler - .compile_promql(r, environment.clone()) - .ok() - }) - .collect(); + let plans: Vec<_> = + super::super::workload_cost::enumerate_exact_and_materialized_candidates(request) + .unwrap() + .into_iter() + .filter_map(|r| { + PhysicalPlanCompiler + .compile_promql(r, environment.clone()) + .ok() + }) + .collect(); let plan = plans.iter().find(|plan| plan.query_plan.entries.values().all(|entry| entry.nodes.values().any(|node| matches!(node, crate::query_plan::QueryPlanNode::Logical { operator: crate::query_plan::residual::ResidualQueryOperator::CurrentSeries { .. }, .. @@ -4165,7 +4133,9 @@ pub(crate) mod tests { entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); } let (request, environment) = snapshot.into_physical_compilation_request().unwrap(); - let candidates = super::super::workload_cost::with_exact_alternative(request).unwrap(); + let candidates = + super::super::workload_cost::enumerate_exact_and_materialized_candidates(request) + .unwrap(); let mut reasons = vec![]; assert!( candidates.into_iter().any(|candidate| { @@ -4551,7 +4521,7 @@ pub(crate) mod tests { #[test] fn hybrid_weighted_topk_installs_only_candidates_and_delegates_filtered_exact_values() { use crate::query_plan::{ - logical::ResidualQueryOperator, ExternalExactInput, ExternalExactOutput, QueryPlanNode, + residual::ResidualQueryOperator, ExternalExactInput, ExternalExactOutput, QueryPlanNode, }; let query = "topk(2, sum by (job) (rate(m[1m])))"; let evidence = TopKMembershipEvidence { @@ -4689,7 +4659,7 @@ pub(crate) mod tests { } #[test] - fn legacy_backend_snapshot_gets_explicit_retained_memory_default_and_alias() { + fn snapshot_uses_retained_memory_default_and_canonical_override() { let source = include_str!("../../../docs/examples/asapquery-planning-snapshot.json"); let snapshot: BackendLocalPlanningInput = serde_json::from_str(source).unwrap(); assert_eq!( @@ -4700,7 +4670,7 @@ pub(crate) mod tests { ); let mut value: Value = serde_json::from_str(source).unwrap(); - value["implementation"]["maxRetainedSummaryBytes"] = json!(123_456); + value["implementation"]["max_retained_summary_bytes"] = json!(123_456); let snapshot: BackendLocalPlanningInput = serde_json::from_value(value).unwrap(); assert_eq!( snapshot @@ -4933,9 +4903,43 @@ pub(crate) mod tests { .any(|node| matches!(node, crate::query_plan::QueryPlanNode::ExactFallback { .. }))); } - /// Distinct range queries retain a per-series HLL selected by Planner. + /// An externally supplied unknown guarantee must not bypass global selection. #[test] - fn distinct_range_compiles_to_partitioned_hll() { + fn supplied_uncertified_readout_is_rejected() { + use asap_aware_mapping::{ + Replacement, ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG, + }; + let mut workload = request("unknown", "distinct_over_time(m[1m])"); + let root = Rc::new( + crate::query_parser::parse_query_expr_canonical( + "distinct_over_time(m[1m])", + AccuracyTarget::Epsilon(0.05), + ) + .unwrap(), + ); + let model = ControlPlaneCostModel::new(AccuracyTarget::Epsilon(0.05)); + workload.queries[0].selected_plan_root = SketchAlgorithmStrategy::new(&model) + .replacements(&TargetSubDAG::new(&root)) + .into_iter() + .find_map(|candidate| { + if !candidate.has_missing_accuracy_evidence() { + return None; + } + match candidate.replacement { + Replacement::Summary(node) => Some(node), + _ => None, + } + }) + .expect("unknown HLL candidate stays inspectable"); + let result = PhysicalPlanCompiler.compile_metricsql(workload, environment(10_000)); + assert!( + matches!(result, Err(CompileError::Query { reason, .. }) if reason.contains("no certified accuracy guarantee")) + ); + } + + /// HLL without a confidence proof remains exact under canonical selection. + #[test] + fn uncertified_distinct_range_remains_exact() { let mut deployment = environment(10_000); deployment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite; deployment.target_collector_ids.clear(); @@ -4958,40 +4962,31 @@ pub(crate) mod tests { let plan = PhysicalPlanCompiler .compile_metricsql(workload, deployment) .unwrap(); - assert_eq!(plan.precompute_plan.materializations.len(), 1); - let materialization = &plan.precompute_plan.materializations[0]; - assert_eq!( - materialization.aggregation_type, - asap_types::AggregationType::HLL - ); - assert_eq!( - materialization.partitioning, - Some(asap_types::sds::PopulationPartitioning::PerEntity) - ); - plan.precompute_plan.validate().unwrap(); - assert!(plan - .query_plan - .entries - .values() - .any(|entry| entry.nodes.values().any(|node| matches!( - node, - crate::query_plan::QueryPlanNode::SummaryEstimate { - query: crate::query_plan::QueryReadout::Cardinality, - .. - } - )))); + assert!(plan.precompute_plan.materializations.is_empty()); + assert!(matches!( + plan.query_plan + .entries + .values() + .next() + .unwrap() + .nodes + .values() + .next() + .unwrap(), + crate::query_plan::QueryPlanNode::ExactFallback { .. } + )); } - /// An unimplemented cardinality family fails admission rather than panicking in an emitter. + /// Mixed execution does not authorize an uncertified cardinality sketch. #[test] - fn unsupported_cardinality_family_fails_admission() { + fn uncertified_cardinality_is_not_materialized() { let mut deployment = environment(10_000); deployment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite; deployment.target_collector_ids.clear(); let mut workload = request("confidence", "distinct_over_time(m[1m])"); workload.allow_mixed_summary_and_exact_execution = true; let result = PhysicalPlanCompiler.compile_metricsql(workload, deployment); - assert!(matches!(result, Err(CompileError::QueryPlan(_)))); + assert!(result.unwrap().precompute_plan.materializations.is_empty()); } #[test] @@ -5162,7 +5157,7 @@ pub(crate) mod tests { // Frozen sketch-bench output exercises the same wire schema on every CI run. #[test] - fn measured_erp_kll_parameters_survive_workload_selection() { + fn measured_erp_without_failure_probability_uses_exact_fallback() { use super::super::erp::{ ErpAccuracyMode, ErpParameterDecision, ErpPlanningInput, ErpRuntimeCapabilities, }; @@ -5226,92 +5221,14 @@ pub(crate) mod tests { ) .unwrap(); - fn contains_measured_kll(node: &SummaryNode) -> bool { - match &node.expr { - SummaryExpr::SummaryAgg { family, child, .. } => { - matches!(family, SummaryFamilyType::Sketch(kind, _) - if matches!(kind.params(), SketchParams::Kll { k: 32 })) - || contains_measured_kll(child) - } - SummaryExpr::SummaryEstimate { summary_input, .. } - | SummaryExpr::ValueOperation { - child: summary_input, - .. - } => contains_measured_kll(summary_input), - _ => false, - } - } - assert!( - contains_measured_kll(&workload.queries[0].selected_plan_root), - "ERP hit was lost before physical compilation: {:#?}", - workload.queries[0].selected_plan_root - ); - let guarantee = workload.queries[0] - .selected_plan_root - .guarantee - .as_ref() - .unwrap(); - assert_eq!(guarantee.failure_probability.evaluate(), None); - assert!(!guarantee.is_exact()); + assert!(matches!( + workload.queries[0].selected_plan_root.expr, + SummaryExpr::KeepPreAsap(_) + )); let plan = PhysicalPlanCompiler .compile_promql(workload, environment(10000)) .unwrap(); - assert_eq!(plan.precompute_plan.materializations[0].parameters["k"], 32); - let empirical_identity = plan.precompute_plan.materializations[0].policy_fingerprint(); - let mut drift = erp.clone(); - drift.distribution = serde_json::json!({"shifted": true}); - let mut unsupported = drift.clone(); - unsupported.runtime.allowed_algorithms = vec![SketchAlgorithm::Hll]; - let mut wrong_implementation = erp.clone(); - wrong_implementation.artifact.records[0].implementation = "oxide".into(); - wrong_implementation.implementation = Some("oxide".into()); - for (policy, accuracy, exact) in [ - (drift, AccuracyTarget::Epsilon(0.06), false), - (wrong_implementation, AccuracyTarget::Epsilon(0.06), false), - ( - erp.clone(), - AccuracyTarget::EpsilonDelta { - epsilon: 0.06, - delta: 0.01, - }, - false, - ), - (unsupported, AccuracyTarget::Epsilon(0.06), true), - ] { - let mut workload = request("q", "quantile_over_time(0.9, m[1m])"); - workload.queries[0].accuracy_target = accuracy.clone(); - let root = Rc::new( - crate::query_parser::parse_query_expr_canonical( - &workload.queries[0].query_string, - accuracy, - ) - .unwrap(), - ); - select_logical_roots_with_error_resource_profiles( - &mut workload.queries, - vec![root], - &workload.topk_membership_evidence_by_query_id, - &workload.exact_composition_costs, - Some(&policy), - ) - .unwrap(); - if exact { - assert!(matches!( - workload.queries[0].selected_plan_root.expr, - SummaryExpr::KeepPreAsap(_) - )); - } else { - assert!(!contains_measured_kll( - &workload.queries[0].selected_plan_root - )); - let plan = PhysicalPlanCompiler - .compile_promql(workload, environment(10000)) - .unwrap(); - let state = &plan.precompute_plan.materializations[0]; - assert!(state.parameters["k"].as_u64().unwrap() > 32); - assert_ne!(state.policy_fingerprint(), empirical_identity); - } - } + assert!(plan.precompute_plan.materializations.is_empty()); } fn measured_exact_composition_rows( @@ -5791,6 +5708,26 @@ pub(crate) mod tests { Query("sum by (service) (sum_over_time(m[1m]) / count_over_time(m[1m]))".into()); entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); let (mut request, _) = snapshot.into_physical_compilation_request().unwrap(); + // Exercise the invalid externally supplied graph, independent of the + // global selector (which now already chooses an exact fallback). + use asap_aware_mapping::ReplacementStrategy; + let candidates = asap_aware_mapping::SketchAlgorithmStrategy::new( + &ControlPlaneCostModel::new(AccuracyTarget::Exact), + ) + .replacements(&asap_aware_mapping::TargetSubDAG::new( + &request.canonical_roots[0], + )); + request.queries[0].selected_plan_root = candidates + .into_iter() + .find_map(|candidate| match candidate.replacement { + asap_aware_mapping::Replacement::Summary(node) + if !matches!(node.expr, SummaryExpr::KeepPreAsap(_)) => + { + Some(node) + } + _ => None, + }) + .expect("invalid summary fixture"); assert!(!matches!( request.queries[0].selected_plan_root.expr, SummaryExpr::KeepPreAsap(_) @@ -5807,8 +5744,13 @@ pub(crate) mod tests { #[test] fn selected_maintenance_dependency_still_requires_a_valid_executable_dag() { - let mut request = request("invalid-dependency", "sum(sum_over_time(m[1m]))"); - let selected = request.queries[0].selected_plan_root.clone(); + let mut request = request("invalid-dependency", "sum_over_time(m[1m])"); + let selected_root = request.queries[0].selected_plan_root.clone(); + let selected = match &selected_root.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => summary_input.clone(), + SummaryExpr::SummaryAgg { .. } => selected_root.clone(), + _ => panic!("expected maintained aggregate fixture"), + }; request.queries[0].selected_plan_root = Rc::new(SummaryNode { expr: SummaryExpr::BinaryOp { timing: planner_types::post_asap::ExecutionTiming::ReadTime, @@ -6822,7 +6764,7 @@ pub(crate) mod tests { // with each operand keeping its own range. #[test] fn composable_binary_summarizes_each_prometheus_filtered_operand() { - use crate::query_plan::{logical::ResidualQueryOperator, QueryPlanNode}; + use crate::query_plan::{residual::ResidualQueryOperator, QueryPlanNode}; let mut snapshot: BackendLocalPlanningInput = serde_json::from_str(include_str!( "../../../docs/examples/asapquery-planning-snapshot.json" )) @@ -7893,7 +7835,13 @@ pub(crate) mod tests { #[test] fn topk_fails_closed_without_membership_evidence() { - assert!(request_with_evidence("q-topk", "topk(5, m)", None).is_err()); + let request = request_with_evidence("q-topk", "topk(5, m)", None).unwrap(); + assert!(request.queries[0] + .selected_plan_root + .guarantee + .as_ref() + .unwrap() + .is_exact()); } #[test] diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index a298e29a..1856672c 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -1454,7 +1454,7 @@ mod tests { )) .unwrap(); let mut query = fixture["query_workload"]["repeating_queries"][3].clone(); - query["query"] = "distinct_over_time(asap_demo_latency_ms[5s])".into(); + query["query"] = "quantile_over_time(0.9,asap_demo_latency_ms[5s])".into(); query["requirements"]["accuracy"] = serde_json::json!({"explicit":{"Epsilon":0.05}}); fixture["query_workload"]["repeating_queries"] = serde_json::json!([query]); let snapshot: crate::physical::compiler::BackendLocalPlanningInput = @@ -1465,16 +1465,28 @@ mod tests { ) .compile_promql() .unwrap(); + // Catalog resolution is independent of Planner selection. Build an + // HLL catalog fixture from a compiled source identity; production + // selection cannot deploy HLL without a known confidence guarantee. + let mut materialization = plan.precompute_plan.materializations[0].clone(); + materialization.aggregation_type = asap_types::AggregationType::HLL; + materialization.parameters = + serde_json::from_value(serde_json::json!({"precision": 14})).unwrap(); + let catalog = asap_types::summary_catalog::SummaryCatalog::from_materializations( + plan.summary_catalog.plan_id, + plan.summary_catalog.plan_version, + &[materialization], + ) + .unwrap(); let (mut policy, mut observed) = online_population_fixture(); - observed.catalog_generation = plan.summary_catalog.reference().unwrap(); - observed.summary_definition_id = - *plan.summary_catalog.materializations.keys().next().unwrap(); + observed.catalog_generation = catalog.reference().unwrap(); + observed.summary_definition_id = *catalog.materializations.keys().next().unwrap(); observed.input_semantics = asap_types::erp_observation::ErpObservationInputSemantics::ScalarSampleValue; policy.observed_populations = Some(observed.clone()); - policy.resolve_population_data_descriptor(Some(&plan.summary_catalog)); - let expected = &plan.summary_catalog.materializations[&observed.summary_definition_id] - .data_descriptor_id; + policy.resolve_population_data_descriptor(Some(&catalog)); + let expected = + &catalog.materializations[&observed.summary_definition_id].data_descriptor_id; assert_eq!( &policy.resolved_data_descriptor.as_ref().unwrap().id, expected diff --git a/control_plane/src/physical/maintained_population.rs b/control_plane/src/physical/maintained_population.rs index e1e6d61c..dd1f856b 100644 --- a/control_plane/src/physical/maintained_population.rs +++ b/control_plane/src/physical/maintained_population.rs @@ -1,8 +1,8 @@ //! Lower typed population operators according to executor membership capabilities. -use super::compiler::{CompileError, PhysicalCompilationRequest, PlanningQuery}; +use super::compiler::{CompileError, PhysicalCompilationRequest, QueryCompilationInput}; use asap_types::query_plan::{ current_series::{SeriesPopulation, SeriesReadout}, - logical::{Grouping, LabelMatch, LabelMatcher, ResidualQueryOperator}, + residual::{Grouping, LabelMatch, LabelMatcher, ResidualQueryOperator}, }; use planner_types::post_asap::{ maintained_population::*, SummaryExpr, SummaryNode, ValueOperation, @@ -42,7 +42,7 @@ pub(super) fn supported(request: &PhysicalCompilationRequest) -> bool { pub(super) fn operator( request: &PhysicalCompilationRequest, - query: &PlanningQuery, + query: &QueryCompilationInput, ) -> Result, CompileError> { let Some((spec, readout)) = selected(&query.selected_plan_root) else { return Ok(None); diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs index afbbd466..571c054c 100644 --- a/control_plane/src/physical/post_asap/cost_model.rs +++ b/control_plane/src/physical/post_asap/cost_model.rs @@ -16,11 +16,12 @@ use asap_aware_mapping::empirical_cost::EmpiricalEvidenceProvider; use asap_aware_mapping::{ CompleteSummaryCandidateEstimate, CostModel, CostProvenance, EvaluationRate, ExactCompositionCostInputs, ExactCompositionCostRequest, Horizon, OperationPlacement, - Realization, SummaryMaintenanceCapabilities, SummaryMaintenanceLifecycleCostInputs, - ValueOperationCapabilities, + Realization, Replacement, ReplacementSubDAG, SummaryMaintenanceCapabilities, + SummaryMaintenanceLifecycleCostInputs, TargetSubDAG, ValueOperationCapabilities, }; use planner_types::post_asap::{ - SketchAlgorithm, SketchParams, SketchQuery, SummaryWindowFramework, + ExecutableOperatorPayload, GroupingStrategy, SketchAlgorithm, SketchParams, SketchQuery, + SummaryFamilyType, SummaryWindowFramework, }; use planner_types::pre_asap::expr_ir::ColumnRef; @@ -31,6 +32,55 @@ use crate::types::AccuracyTarget; use planner_types::pre_asap::AggIntent; use serde::{Deserialize, Serialize}; +/// A local state-footprint estimate, never a complete deployment quote. +#[derive(Debug, Serialize)] +pub struct CandidateCostEstimate { + pub value: f64, + pub unit: &'static str, + pub model: &'static str, + pub source: &'static str, + pub erp_record_ids: Vec, +} + +fn analytical_state_bytes(family: &SummaryFamilyType) -> Option { + use asap_types::AggregationType as A; + use planner_types::post_asap::ExactKind; + let (aggregation, params) = match family { + SummaryFamilyType::ExactAggregate(kind, _) => ( + match kind { + ExactKind::Sum | ExactKind::Count => A::Sum, + ExactKind::Min => A::Min, + ExactKind::Max => A::Max, + ExactKind::Increase | ExactKind::Rate | ExactKind::IRate => A::Increase, + }, + std::collections::HashMap::new(), + ), + SummaryFamilyType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance) => { + let aggregation = match kind.algorithm() { + SketchAlgorithm::DDSketch => A::DDSketch, + SketchAlgorithm::Kll => A::DatasketchesKLL, + SketchAlgorithm::Hll => A::HLL, + SketchAlgorithm::Cms => A::CountMinSketch, + SketchAlgorithm::CountSketch => A::CountSketch, + SketchAlgorithm::CmsWithHeap => A::CountMinSketchWithHeap, + SketchAlgorithm::CountSketchWithHeap => A::CountSketchWithHeap, + SketchAlgorithm::UnivMon => A::UnivMon, + SketchAlgorithm::Kmv | SketchAlgorithm::Theta => return None, + }; + let params = super::super::compiler::sketch_params_json(kind.params()) + .as_object()? + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect(); + (aggregation, params) + } + // A shared grid needs its own population/layout model; an independent + // state's measurement is not a measurement of that grid. + _ => return None, + }; + Some(super::super::compiler::estimated_state_bytes(&aggregation, ¶ms) as f64) +} + /// One measured execution profile for an exact operator composed with a /// maintained summary. Values use CPU nanoseconds so every term in Planner's /// recurring-cost formula has the same physical unit. Peak memory is retained @@ -120,6 +170,39 @@ pub struct ControlPlaneCostModel { } impl ControlPlaneCostModel { + pub fn candidate_cost_estimate( + &self, + candidate: &ReplacementSubDAG, + ) -> Option { + let Replacement::Summary(root) = &candidate.replacement else { + // Exact compositions have a separate measured rate model. Raw + // rewrites have no retained-state estimate in this model. + return None; + }; + let dag = planner_types::post_asap::compile_executable_dag(root).ok()?; + let mut value = 0.0; + let mut states = 0; + for node in &dag.nodes { + let family = match &node.payload { + ExecutableOperatorPayload::SummaryAgg { family, .. } + | ExecutableOperatorPayload::SummaryJoin { family, .. } => family, + _ => continue, + }; + states += 1; + value += analytical_state_bytes(family)?; + } + if states == 0 || !value.is_finite() { + return None; + } + Some(CandidateCostEstimate { + value, + unit: "bytes_per_state_partition", + model: "backend_state_footprint_v1", + source: "analytical", + erp_record_ids: vec![], + }) + } + pub fn new(workload_accuracy: AccuracyTarget) -> Self { Self { workload_accuracy, @@ -427,6 +510,15 @@ fn intent_accuracy(intent: &AggIntent) -> AccuracyTarget { } impl CostModel for ControlPlaneCostModel { + fn candidate_cost( + &self, + candidate: &ReplacementSubDAG, + _target: &TargetSubDAG<'_>, + ) -> Option { + self.candidate_cost_estimate(candidate) + .map(|estimate| Cost(estimate.value)) + } + fn value_operation_capabilities(&self) -> ValueOperationCapabilities { ValueOperationCapabilities { read_time: true, @@ -777,6 +869,14 @@ impl ForcedFamilyCostModel { } impl CostModel for ForcedFamilyCostModel { + fn candidate_cost( + &self, + candidate: &ReplacementSubDAG, + target: &TargetSubDAG<'_>, + ) -> Option { + self.inner.candidate_cost(candidate, target) + } + fn rank_candidates( &self, intent: &AggIntent, diff --git a/control_plane/src/physical/post_asap/deployment_expr.rs b/control_plane/src/physical/post_asap/deployment_expr.rs index 4e1ba95d..4012277e 100644 --- a/control_plane/src/physical/post_asap/deployment_expr.rs +++ b/control_plane/src/physical/post_asap/deployment_expr.rs @@ -172,7 +172,14 @@ mod tests { having: None, child: Rc::new(windowed_scan()), }; - let node = crate::planner_selection::select_summary_default(&q).expect("implements"); + let node = crate::planner_selection::select_query( + &q, + &crate::physical::post_asap::cost_model::ForcedFamilyCostModel::new( + crate::types::AccuracyTarget::Epsilon(0.01), + planner_types::post_asap::SketchAlgorithm::Kll, + ), + ) + .expect("implements"); let e = PhysicalExpr::committed(node); match e { PhysicalExpr::Committed(PostAsapPlan::Summary(node)) => match &node.expr { diff --git a/control_plane/src/physical/post_asap/lower.rs b/control_plane/src/physical/post_asap/lower.rs index 5d168c89..cdff8f35 100644 --- a/control_plane/src/physical/post_asap/lower.rs +++ b/control_plane/src/physical/post_asap/lower.rs @@ -1,30 +1,5 @@ -//! L3 → L4/L5 lowering — `QueryExpr` walk that selects candidates from -//! `SketchAlgorithmStrategy::replacements` via `planner_selection::select_summary`, with -//! `crate::physical::post_asap::cost_model::ControlPlaneCostModel` plugged in -//! for family selection + parameter sizing. -//! -//! Per `control_plane/docs/design.md` §6: "the optimizer's job is to -//! selectively replace logical aggregates / joins with their sketch-bound -//! variants when a binding rule fires; everything else stays inside -//! `Logical(…)`." -//! -//! Two node shapes are rewritten *before* selecting a candidate, because -//! the upstream strategy can produce summaries this deployment's data plane -//! doesn't (or, deliberately, shouldn't) serve — not something the -//! `CostModel` hook can reach, since the decision of *whether* to call -//! into `rank_candidates`/`size_params` at all is made before the -//! `CostModel` is ever consulted. See each helper's docs for the specific -//! reason. -//! -//! `AggIntent::Extension` (the `Frequency` point-query) needs no such -//! pre-pass anymore: `ControlPlaneCostModel::realize_extension`/ -//! `readout_extension` (ASAPController#150) now realize it as a real -//! `CountSketch`, so the catch-all arm below selects it like any other intent. -//! `AggIntent::TopK { accuracy: Exact }` is the one remaining case left to -//! fall through to the strategy's `KeepPreAsap` fallback -//! unchanged — a genuine, still-open `asap-plan` coverage gap (filed -//! upstream — see ASAPController#151), not something this deployment -//! should route around locally. +//! Query binding delegates selection to Planner's costed workload search. +//! Backend-specific rate normalization remains part of the physical binding. #![allow(dead_code)] @@ -139,13 +114,13 @@ fn bind_recursive( ) => { Ok(PostAsapPlan::Summary( - crate::planner_selection::select_summary(expr, cost_model)?, + crate::planner_selection::select_query(expr, cost_model)?, )) } _ => { let rewritten = rewrite_rate_to_increase(expr); - let node = crate::planner_selection::select_summary(&rewritten, cost_model)?; + let node = crate::planner_selection::select_query(&rewritten, cost_model)?; Ok(PostAsapPlan::Summary(node)) } } diff --git a/control_plane/src/physical/post_asap/tests.rs b/control_plane/src/physical/post_asap/tests.rs index 3f12dec6..e2eec8b9 100644 --- a/control_plane/src/physical/post_asap/tests.rs +++ b/control_plane/src/physical/post_asap/tests.rs @@ -138,7 +138,7 @@ fn bind_kll_quantile_basic() { let expr = agg_quantile(0.99, AccuracyTarget::Epsilon(0.01)); let cost_model = ForcedFamilyCostModel::new(AccuracyTarget::Epsilon(0.01), SketchAlgorithm::Kll); - let node = crate::planner_selection::select_summary(&expr, &cost_model) + let node = crate::planner_selection::select_query(&expr, &cost_model) .expect("KLL should bind a Quantile{0.99, ε=0.01}"); match &node.expr { SummaryExpr::SummaryEstimate { @@ -167,7 +167,7 @@ fn bind_ddsketch_quantile_basic() { let expr = agg_quantile(0.99, AccuracyTarget::Epsilon(0.01)); let cost_model = ForcedFamilyCostModel::new(AccuracyTarget::Epsilon(0.01), SketchAlgorithm::DDSketch); - let node = crate::planner_selection::select_summary(&expr, &cost_model) + let node = crate::planner_selection::select_query(&expr, &cost_model) .expect("DDSketch should bind a Quantile{0.99, ε=0.01}"); match &node.expr { SummaryExpr::SummaryEstimate { @@ -276,52 +276,33 @@ fn topk_binding_family(bound: &PhysicalExpr) -> (SketchAlgorithm, u32, u32) { } } -/// (a) A **loose-recall** top-k (any non-exact accuracy target) binds the -/// cheap **CMS-with-heap** family — the Fig-12 cost-gap fix. The old rule -/// hard-bound the ~66×-more-expensive CountSketch here. +/// A loose TopK target still needs membership evidence to select a sketch. #[test] -fn bind_cms_topk_loose_recall_picks_cms_heap() { +fn uncertified_topk_keeps_exact_execution() { let acc = AccuracyTarget::EpsilonDelta { epsilon: 0.01, delta: 0.001, }; let expr = agg_topk(10, acc.clone()); - assert!(bind_query_expr(&expr, acc).is_err()); + assert!(query_is_exact(&committed_node( + bind_query_expr(&expr, acc).unwrap() + ))); } -/// (b) A **tight / exact-recall** top-k binds the unbiased -/// **CountSketch-with-heap** — the family that supports exact rank / -/// signed estimates. -/// -/// NOTE — behavior change forced by the new realization pass, not just a rename: -/// the old fixture used `AggIntent::TopK{accuracy: Exact}` (the intent's -/// OWN accuracy) to signal "tight/exact-recall". Under -/// `asap_aware_mapping::replacement::realizations_for_intent`, the per-intent -/// summary-vs-exact boundary decision checks the intent's own `accuracy` -/// field FIRST: `TopK{accuracy: Exact}` now declines to bind at all -/// (`SummaryExpr::KeepPreAsap`) rather than reaching the cost model's -/// family-selection logic at all — see `topk_exact_accuracy_declines_to_bind` -/// above (a REAL, accepted behavior change — ASAPController#151 — per -/// this migration's design notes, not a bug to route around). "Tight -/// recall" (→ CountSketchWithHeap) is still live logic in -/// `ControlPlaneCostModel::topk_family_order` — it fires off the -/// WORKLOAD-level accuracy (not the intent's own) being `Exact`, which -/// still lets the intent itself bind. +/// An exact workload target cannot accept uncertified TopK membership. #[test] -fn bind_cms_topk_tight_recall_picks_countsketch() { +fn exact_topk_keeps_exact_execution() { // Intent requests a normal (non-exact) rank so binding still // happens; the workload-level policy demands exact recall. let expr = agg_topk(10, AccuracyTarget::Epsilon(0.01)); - assert!(bind_query_expr(&expr, AccuracyTarget::Exact).is_err()); + assert!(query_is_exact(&committed_node( + bind_query_expr(&expr, AccuracyTarget::Exact).unwrap() + ))); } -/// (c) The chosen family is the **cost-minimal one that meets the recall -/// SLA**, per the `physical::deployment_cost::wire` table — the same "min cost s.t. -/// SLA" the oracle uses. Loose → both families clear the bar → cheapest -/// (CMS, ~4 KB) wins; the CountSketch alternative (~250 KB) is ~66× -/// costlier. +/// A cheaper sketch never substitutes for missing membership evidence. #[test] -fn bind_cms_topk_picks_cost_min_meeting_sla() { +fn cheap_topk_does_not_bypass_membership_evidence() { use crate::physical::deployment_cost::wire::WireCostTable; let table = WireCostTable::default(); let cms = table.for_algorithm(&SketchAlgorithm::Cms).per_flush(); @@ -341,11 +322,13 @@ fn bind_cms_topk_picks_cost_min_meeting_sla() { // Loose recall → the planner must land on the cost-min family (CMS). let acc = AccuracyTarget::Epsilon(0.01); - assert!(bind_query_expr(&agg_topk(10, acc.clone()), acc).is_err()); + assert!(query_is_exact(&committed_node( + bind_query_expr(&agg_topk(10, acc.clone()), acc).unwrap() + ))); } #[test] -fn bind_hll_cardinality_basic() { +fn uncertified_hll_keeps_exact_execution() { let expr = QueryExpr::Aggregate { reduction: Reduction::PerEntity, measures: vec![AggIntent::Cardinality { @@ -357,35 +340,7 @@ fn bind_hll_cardinality_basic() { child: Rc::new(windowed_scan()), }; let bound = bind_query_expr(&expr, AccuracyTarget::Epsilon(0.01)).expect("no error"); - match bound { - PhysicalExpr::Committed(PostAsapPlan::Summary(node)) => match &node.expr { - SummaryExpr::SummaryEstimate { - query, - summary_input, - } => { - assert!(matches!(query, SketchQuery::Cardinality)); - match &summary_input.expr { - SummaryExpr::SummaryAgg { family, .. } => match family { - SummaryFamilyType::Sketch(kind, _) - if kind.algorithm() == &SketchAlgorithm::Hll => - { - let SketchParams::Hll { precision } = kind.params() else { - panic!("HLL algorithm has mismatched params") - }; - assert!( - *precision >= 12, - "ε=0.01 should land on at least precision 12 (~1.6%) per the rung table" - ); - } - other => panic!("expected Hll family, got {other:?}"), - }, - other => panic!("expected SummaryAgg, got {other:?}"), - } - } - other => panic!("expected SummaryEstimate, got {other:?}"), - }, - other => panic!("expected Committed(Summary(_)), got {other:?}"), - } + assert!(query_is_exact(&committed_node(bound))); } #[test] @@ -735,17 +690,30 @@ fn phase_b_e2e_rate_falls_through_to_logical() { ); } -/// The legacy single-expression binder cannot choose a frequency sketch for -/// value-ranked `topk(sum(rate(...)))` without membership evidence. The -/// workload planner handles this query as query-time Sort+Limit over its -/// recursively planned child; its coverage lives in `query_plan::residual`. +/// Value-ranked TopK candidates with missing membership evidence remain +/// inspectable but cannot be reported as certified selections. #[test] -fn phase_b_legacy_topk_requires_membership_evidence() { +fn value_ranked_topk_without_membership_evidence_is_not_certified() { let query = "topk(10, sum by (instance) (rate(http_requests_total[5m])))"; let accuracy = AccuracyTarget::Epsilon(0.05); let expr = crate::query_parser::parse_query_expr_canonical(query, accuracy.clone()) .expect("TopK parses"); - assert!(bind_query_expr(&expr, accuracy).is_err()); + let (_, trace) = crate::planner_selection::select_workload_with_accuracy_model_and_trace( + vec![(0, std::rc::Rc::new(expr))], + accuracy.clone(), + &crate::physical::post_asap::cost_model::ControlPlaneCostModel::new(accuracy), + &asap_aware_mapping::NoAccuracyEvidence, + &asap_aware_mapping::DefaultAccuracyModel, + ) + .unwrap(); + assert!(trace["groups"] + .as_array() + .unwrap() + .iter() + .flat_map(|group| group["candidates"].as_array().unwrap()) + .any( + |candidate| candidate["accuracy_status"] == "unknown" && candidate["selected"] == false + )); } /// Archive-only routing through the full L1→L3→L4 pipeline. Asserts the @@ -909,3 +877,16 @@ fn topk_exact_accuracy_declines_to_bind() { other => panic!("expected Committed(Summary(_)), got {other:?}"), } } + +fn committed_node(bound: PhysicalExpr) -> Rc { + let PhysicalExpr::Committed(PostAsapPlan::Summary(node)) = bound else { + panic!("expected committed query") + }; + node +} + +fn query_is_exact(node: &SummaryNode) -> bool { + node.guarantee + .as_ref() + .is_some_and(|guarantee| guarantee.is_exact()) +} diff --git a/control_plane/src/physical/publication.rs b/control_plane/src/physical/publication.rs index b12ac10c..6bd088f3 100644 --- a/control_plane/src/physical/publication.rs +++ b/control_plane/src/physical/publication.rs @@ -15,10 +15,3 @@ impl CompiledPhysicalPlan { Ok(artifact) } } - -impl CompiledPhysicalPlan { - #[deprecated(note = "Use to_publication_artifact")] - pub fn publication(&self) -> Result { - self.to_publication_artifact() - } -} diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index 5816ce8e..71f7bfa2 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -29,9 +29,9 @@ pub struct CostComponentDemand { pub implementation: Value, /// `horizon` includes all work in the manifest's source/time scope; /// `query_evaluation` is one execution of this bound query operator. - #[serde(rename = "unit", alias = "pricing_basis")] + #[serde(rename = "unit")] pub pricing_basis: String, - #[serde(rename = "multiplicity", alias = "occurrences_per_horizon")] + #[serde(rename = "multiplicity")] pub occurrences_per_horizon: f64, } @@ -44,7 +44,7 @@ pub struct WorkloadCostManifest { pub capability_snapshot_id: String, pub backend_compat: String, pub horizon_seconds: f64, - /// Canonical roots, requirements and demand must match across alternatives. + /// Canonical roots, requirements and demand must match across candidates. pub workload: BTreeMap, pub components: BTreeMap, } @@ -79,11 +79,11 @@ pub struct WorkloadCostEvidence { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] /// A candidate diagnostic can precede pricing or record compilation failure. pub struct CandidatePlanEvaluation { - #[serde(rename = "alternative_id", alias = "candidate_id")] + #[serde(rename = "candidate_id")] pub candidate_id: Option, #[serde(default)] pub logical_root_ids: Vec, - #[serde(rename = "physical_alternative_id", alias = "physical_candidate_id")] + #[serde(rename = "physical_candidate_id")] pub physical_candidate_id: Option, pub identity_unavailable_reason: Option, #[serde(default)] @@ -95,31 +95,28 @@ pub struct CandidatePlanEvaluation { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct MaterializationSearchCoverage { - #[serde(rename = "eligible_leaves", alias = "eligible_materialization_count")] + #[serde(rename = "eligible_leaves")] pub eligible_materialization_count: usize, - #[serde( - rename = "enumerated_local_masks", - alias = "enumerated_candidate_key_sets" - )] + #[serde(rename = "enumerated_local_masks")] pub enumerated_candidate_key_sets: usize, pub exhaustive: bool, - #[serde(rename = "scope", alias = "search_scope")] + #[serde(rename = "scope")] pub search_scope: CandidateSearchScope, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct CandidatePlanSelectionReport { #[serde(default)] - #[serde(rename = "logical_selection", alias = "planner_selection_trace")] + #[serde(rename = "logical_selection")] pub planner_selection_trace: Vec, - #[serde(default, alias = "index_search_coverage")] + #[serde(default)] pub materialization_search_coverage: Option, pub data_snapshot_id: String, pub model_version: String, pub selected_plan_id: u64, pub selected_manifest: WorkloadCostManifest, pub component_costs: BTreeMap, - #[serde(rename = "alternatives", alias = "candidate_evaluations")] + #[serde(rename = "candidates")] pub candidate_evaluations: Vec, } @@ -470,7 +467,7 @@ impl WorkloadCostEvidence { } } -fn alternative_description(candidate: &PhysicalCompilationRequest) -> CandidatePlanEvaluation { +fn candidate_description(candidate: &PhysicalCompilationRequest) -> CandidatePlanEvaluation { let root_ids = candidate .queries .iter() @@ -519,7 +516,7 @@ fn compile_candidate_for_pricing( ), Box, > { - let mut description = alternative_description(&candidate); + let mut description = candidate_description(&candidate); let queries = candidate.queries.clone(); let compiled = super::realization::RealizationProvider::compile( &super::realization::ExistingRealizations, @@ -606,7 +603,7 @@ pub fn compile_candidates_for_pricing( } /// Compare complete Planner-authorized forests after binding. Infeasible or -/// uncosted alternatives are retained as unavailable, never assigned zero. +/// uncosted candidates are retained as unavailable, never assigned zero. pub fn select_lowest_cost_candidate( candidates: Vec, env: PhysicalDeploymentContext, @@ -642,7 +639,7 @@ fn select_candidates( evidence.validate(&env)?; if candidates.is_empty() || candidates.len() > 64 { return Err(invalid( - "candidate inventory must contain 1..=64 alternatives", + "candidate inventory must contain 1..=64 candidates", )); } let candidate_key_sets: BTreeSet<_> = candidates @@ -686,9 +683,7 @@ fn select_candidates( .as_ref() .is_some_and(|previous| previous != &scope) { - return Err(invalid( - "alternatives describe different workloads/horizons", - )); + return Err(invalid("candidates describe different workloads/horizons")); } comparison_workload = Some(scope); match super::realization::RealizationProvider::price( @@ -713,9 +708,9 @@ fn select_candidates( } } let (_, mut plan, selected_manifest, component_costs) = best.ok_or_else(|| { - CompileError::Alternatives( + CompileError::Candidates( json!({"status": "all_infeasible", "logical_selection": planner_selection_trace, - "alternatives": candidate_evaluations}), + "candidates": candidate_evaluations}), ) })?; // Exactly the winner retained by the existing strict-less-than selector. @@ -739,11 +734,11 @@ pub fn enumerate_exact_and_materialized_candidates( request: PhysicalCompilationRequest, ) -> Result, CompileError> { let already_selected = super::maintained_population::supported(&request); - let mut alternatives = materialization_alternatives(request)?; + let mut candidates = materialization_candidates(request)?; if already_selected { - return Ok(alternatives); + return Ok(candidates); } - let roots: Vec<_> = alternatives + let roots: Vec<_> = candidates .last() .expect("exact alternative") .queries @@ -755,7 +750,7 @@ pub fn enumerate_exact_and_materialized_candidates( .collect(); let strategy = asap_aware_mapping::maintained_population::MaintainedPopulationStrategy::new(&roots); - let candidates: Vec<_> = roots + let maintained_roots: Vec<_> = roots .iter() .map(|root| { strategy @@ -763,19 +758,19 @@ pub fn enumerate_exact_and_materialized_candidates( .filter(|node| super::maintained_population::supported_node(node)) }) .collect(); - if candidates.iter().any(Option::is_some) { + if maintained_roots.iter().any(Option::is_some) { // Current-series rules are compatible with window summaries in other // workload roots. Preserve each priced temporal alternative and mask. - let maintained: Vec<_> = alternatives + let maintained: Vec<_> = candidates .iter() .map(|alternative| { let mut candidate = alternative.clone(); - for (query, selected) in candidate.queries.iter_mut().zip(&candidates) { + for (query, selected) in candidate.queries.iter_mut().zip(&maintained_roots) { if let Some(selected) = selected { query.selected_plan_root = std::rc::Rc::clone(selected); } } - if candidates.iter().all(Option::is_some) { + if maintained_roots.iter().all(Option::is_some) { candidate.allow_mixed_summary_and_exact_execution = false; candidate.enabled_materialization_keys = None; } @@ -783,7 +778,7 @@ pub fn enumerate_exact_and_materialized_candidates( }) .collect(); for candidate in maintained { - if !alternatives.iter().any(|existing| { + if !candidates.iter().any(|existing| { existing.allow_mixed_summary_and_exact_execution == candidate.allow_mixed_summary_and_exact_execution && existing.enabled_materialization_keys @@ -794,14 +789,14 @@ pub fn enumerate_exact_and_materialized_candidates( .zip(&candidate.queries) .all(|(a, b)| a.selected_plan_root == b.selected_plan_root) }) { - alternatives.push(candidate); + candidates.push(candidate); } } } - Ok(alternatives) + Ok(candidates) } -fn materialization_alternatives( +fn materialization_candidates( request: PhysicalCompilationRequest, ) -> Result, CompileError> { let mut exact = request.clone(); @@ -868,37 +863,6 @@ fn materialization_alternatives( } } -// Compatibility imports; new callers use the domain names above. -#[deprecated(note = "Use CandidatePlanEvaluation")] -pub use CandidatePlanEvaluation as AlternativeCost; -#[deprecated(note = "Use CandidatePlanSelectionReport")] -pub use CandidatePlanSelectionReport as WorkloadCostComparison; -#[deprecated(note = "Use CostComponentDemand")] -pub use CostComponentDemand as CostDemand; - -#[deprecated(note = "Use enumerate_exact_and_materialized_candidates")] -pub use enumerate_exact_and_materialized_candidates as with_exact_alternative; -#[deprecated(note = "Use select_lowest_cost_candidate")] -pub use select_lowest_cost_candidate as select; -#[deprecated(note = "Use select_lowest_cost_metricsql_candidate")] -pub use select_lowest_cost_metricsql_candidate as select_metricsql; -#[deprecated(note = "Use compile_candidates_for_pricing with QueryFrontend")] -pub fn prepare_manifests( - candidates: Vec, - env: PhysicalDeploymentContext, - metricsql: bool, -) -> (Vec, Vec) { - compile_candidates_for_pricing( - candidates, - env, - if metricsql { - super::compiler::QueryFrontend::MetricsQl - } else { - super::compiler::QueryFrontend::PromQl - }, - ) -} - #[cfg(test)] mod tests { use super::super::compiler::BackendLocalPlanningInput; @@ -914,74 +878,33 @@ mod tests { snapshot } - // New input aliases must produce the same candidate identities and manifests - // while serialization continues to serve existing evidence producers. + /// Input uses one wire contract; removed aliases are not silently accepted. #[test] - fn renamed_inputs_preserve_candidate_manifests_and_wire_names() { - let legacy = serde_json::to_value(fixture()).unwrap(); - assert!(legacy.get("snapshot_version").is_some()); - assert!(legacy.get("physical_inputs").is_none()); - let mut renamed = legacy.clone(); - let root = renamed.as_object_mut().unwrap(); - let version = root.remove("snapshot_version").unwrap(); - root.insert("schema_version".into(), version); - let inputs = root.remove("implementation").unwrap(); - // Upstream window planning now consumes a cost model; removed default - // window fields are no longer part of the naming compatibility contract. - assert!(inputs.get("window_cost_model").is_some()); - assert!(inputs.get("window_implementation_id").is_none()); - assert!(inputs.get("implementation_cost").is_none()); - root.insert("physical_inputs".into(), inputs); - let environment = root - .get_mut("environment") - .unwrap() - .as_object_mut() - .unwrap(); - let collectors = environment.remove("collector_ids").unwrap(); - environment.insert("target_collector_ids".into(), collectors); - - let old: BackendLocalPlanningInput = serde_json::from_value(legacy.clone()).unwrap(); - let new: BackendLocalPlanningInput = serde_json::from_value(renamed).unwrap(); - assert_eq!(old, new); - assert_eq!(serde_json::to_value(&new).unwrap(), legacy); - // Shared publication fields already had domain names: renaming the - // streaming accessor must not change their wire keys or catalog hash. - let (request, environment) = new.clone().into_physical_compilation_request().unwrap(); - let plan = PhysicalPlanCompiler - .compile_promql(request, environment) - .unwrap(); - for document in [ - serde_json::to_value(&plan.summary_catalog).unwrap(), - serde_json::to_value(&plan.precompute_plan).unwrap(), + fn snapshot_rejects_removed_input_aliases() { + let value = serde_json::to_value(fixture()).unwrap(); + let decoded: BackendLocalPlanningInput = serde_json::from_value(value.clone()).unwrap(); + assert_eq!(serde_json::to_value(decoded).unwrap(), value); + for (canonical, removed) in [ + ("snapshot_version", "schema_version"), + ("implementation", "physical_inputs"), ] { - assert!(document.get("materializations").is_some()); - assert!(document.get("get_all_aggregation_configs").is_none()); + let mut invalid = value.clone(); + let field = invalid.as_object_mut().unwrap().remove(canonical).unwrap(); + invalid[removed] = field; + assert!(serde_json::from_value::(invalid).is_err()); } - let compile = |input: BackendLocalPlanningInput| { - let (request, environment) = input.into_physical_compilation_request().unwrap(); - compile_candidates_for_pricing( - enumerate_exact_and_materialized_candidates(request).unwrap(), - environment, - super::super::compiler::QueryFrontend::PromQl, - ) - }; - let old_candidates = compile(old); - let new_candidates = compile(new); - assert!(!old_candidates.0.is_empty()); - assert_eq!(old_candidates, new_candidates); } - // A renamed demand remains readable by old quote providers, including - // fractional recurrence; missing and future statuses keep round-tripping. + // Fractional demand and unavailable candidate costs preserve the current contract. #[test] - fn demand_and_evaluation_keep_legacy_wire_contracts() { + fn demand_and_evaluation_use_candidate_wire_contracts() { let old = json!({"implementation": {"op": "read"}, "unit": "query_evaluation", "multiplicity": 2.5}); let demand: CostComponentDemand = serde_json::from_value(old.clone()).unwrap(); assert_eq!(demand.pricing_basis, "query_evaluation"); assert_eq!(demand.occurrences_per_horizon, 2.5); assert_eq!(serde_json::to_value(demand).unwrap(), old); let row = json!({ - "alternative_id": null, "physical_alternative_id": null, + "candidate_id": null, "physical_candidate_id": null, "identity_unavailable_reason": null, "plan_id": null, "total_cost": null, "unavailable_reason": null }); @@ -989,11 +912,11 @@ mod tests { assert_eq!(evaluation.status, CandidateEvaluationStatus::Unspecified); let encoded = serde_json::to_value(evaluation).unwrap(); assert_eq!(encoded["status"], ""); - assert!(encoded.get("alternative_id").is_some()); - assert!(encoded.get("candidate_id").is_none()); + assert!(encoded.get("candidate_id").is_some()); + assert!(encoded.get("alternative_id").is_none()); } - // IDs describe semantics; activation/version changes do not create new alternatives. + // IDs describe semantics; activation/version changes do not create new candidates. #[test] fn explain_identity_is_stable_across_activations_and_distinguishes_native() { let (request, mut env) = fixture().into_physical_compilation_request().unwrap(); @@ -1050,15 +973,15 @@ mod tests { let (candidates, env, mut evidence) = quoted(); let count = candidates.len(); evidence.quotes.clear(); - let CompileError::Alternatives(report) = + let CompileError::Candidates(report) = select_lowest_cost_candidate(candidates, env, &evidence).unwrap_err() else { panic!("expected structured all-infeasible report") }; assert_eq!(report["status"], "all_infeasible"); - let alternatives = report["alternatives"].as_array().unwrap(); - assert_eq!(alternatives.len(), count); - assert!(alternatives + let candidates = report["candidates"].as_array().unwrap(); + assert_eq!(candidates.len(), count); + assert!(candidates .iter() .all(|item| item["status"] == "evidence_missing")); assert!(!report["logical_selection"].as_array().unwrap().is_empty()); diff --git a/control_plane/src/planner_selection.rs b/control_plane/src/planner_selection.rs index b536460c..37e17fd7 100644 --- a/control_plane/src/planner_selection.rs +++ b/control_plane/src/planner_selection.rs @@ -6,10 +6,11 @@ use std::rc::Rc; +use crate::physical::post_asap::cost_model::ControlPlaneCostModel; use crate::types::AccuracyTarget; use asap_aware_mapping::{ - AccuracyBudgetAllocator, AccuracyEvidenceProvider, AccuracyModel, CostModel, Replacement, - ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG, + AccuracyEvidenceProvider, AccuracyModel, CostModel, Replacement, ReplacementStrategy, + SketchAlgorithmStrategy, TargetSubDAG, }; use planner_types::post_asap::{ SummaryExpr, SummaryFamilyType, SummaryField, SummaryNode, SummarySchema, @@ -231,29 +232,26 @@ pub fn keep_pre_asap(expr: &QueryExpr) -> Result, SelectionError })) } -/// Select the first legal candidate after the supplied deployment cost model -/// has ranked Planner's exhaustive candidate set. -pub fn select_summary( +/// Select a complete query through the same costed Planner search as workloads. +/// Per-operator requirements remain in the canonical expression. +pub fn select_query( expr: &QueryExpr, cost_model: &dyn CostModel, ) -> Result, SelectionError> { - let root = Rc::new(expr.clone()); - let strategy = SketchAlgorithmStrategy::new(cost_model); - let candidate = strategy - .replacements(&TargetSubDAG::new(&root)) - .into_iter() - .next() - .ok_or(SelectionError::NoLegalCandidate)?; - match candidate.replacement { - Replacement::Summary(node) => Ok(node), - Replacement::Rewrite(_) | Replacement::ExactComposition(_) => { - Err(SelectionError::UnexpectedRewrite) - } - } + select_query_with_models( + expr, + cost_model, + &asap_aware_mapping::DefaultAccuracyModel, + &asap_aware_mapping::NoAccuracyEvidence, + ) } -pub fn select_summary_default(expr: &QueryExpr) -> Result, SelectionError> { - select_summary(expr, &asap_aware_mapping::DefaultCostModel) +#[cfg(test)] +pub(crate) fn plan_test_query(expr: &QueryExpr) -> Result, SelectionError> { + select_query( + expr, + &ControlPlaneCostModel::new(AccuracyTarget::Epsilon(0.01)), + ) } #[cfg(test)] @@ -300,7 +298,15 @@ pub fn select_workload_with_accuracy_model( evidence: &dyn AccuracyEvidenceProvider, accuracy_model: &dyn AccuracyModel, ) -> Result)>, SelectionError> { - select_workload_impl(roots, accuracy, cost_model, evidence, accuracy_model, None) + select_workload_impl( + roots, + accuracy, + cost_model, + evidence, + accuracy_model, + None, + None, + ) } /// Return the candidate ranking and committed choices from the same search @@ -308,7 +314,7 @@ pub fn select_workload_with_accuracy_model( pub fn select_workload_with_accuracy_model_and_trace( roots: Vec<(usize, Rc)>, accuracy: AccuracyTarget, - cost_model: &dyn CostModel, + cost_model: &ControlPlaneCostModel, evidence: &dyn AccuracyEvidenceProvider, accuracy_model: &dyn AccuracyModel, ) -> Result<(Vec<(usize, Rc)>, serde_json::Value), SelectionError> { @@ -320,6 +326,7 @@ pub fn select_workload_with_accuracy_model_and_trace( evidence, accuracy_model, Some(&mut trace), + Some(cost_model), )?; Ok((selected, trace)) } @@ -386,6 +393,7 @@ fn select_workload_impl( evidence: &dyn AccuracyEvidenceProvider, accuracy_model: &dyn AccuracyModel, mut trace: Option<&mut serde_json::Value>, + backend_cost_model: Option<&ControlPlaneCostModel>, ) -> Result)>, SelectionError> { let strategies = replacement_strategies(cost_model, evidence, accuracy_model); let space = asap_aware_mapping::search_workload_with_targets( @@ -401,8 +409,36 @@ fn select_workload_impl( let groups = space.cost_sorted(cost_model).iter().enumerate().map(|(index, group)| { let chosen = selection.target_selections().find(|selected| Rc::ptr_eq(selected.target, group.target)) .and_then(|selected| selected.chosen); - let candidates = group.candidates.iter().zip(&group.costs).enumerate() - .map(|(rank, (candidate, cost))| serde_json::json!({ + let target = TargetSubDAG::with_consumer_count(group.target, group.consumer_count); + let candidates = group.candidates.iter().enumerate() + .map(|(rank, candidate)| { + let candidate_cost = cost_model.candidate_cost(candidate, &target); + let accuracy_status = if candidate.has_missing_accuracy_evidence() { + "unknown" + } else { + "known" + }; + let runtime_support_status = match candidate.runtime_support_evidence(cost_model) { + Some(true) => "supported", + Some(false) => "unsupported", + None => "unknown_pending_backend_binding", + }; + let decision_reason = if candidate.has_missing_accuracy_evidence() { + "missing_accuracy_evidence" + } else if candidate.runtime_support_evidence(cost_model) == Some(false) { + "unsupported_runtime_operation" + } else if chosen.is_some_and(|chosen| std::ptr::eq(chosen, *candidate)) { + "planner_selected_pending_backend_binding" + } else if candidate_cost.is_none() { + "missing_comparable_cost" + } else { + "not_selected_by_planner" + }; + let guarantee = match &candidate.replacement { + Replacement::Summary(node) => node.guarantee.as_ref(), + _ => None, + }; + serde_json::json!({ "rank": rank, "candidate_id": replacement_identity(group.target, &candidate.replacement, &accuracy), "status": if chosen.is_some_and(|chosen| std::ptr::eq(chosen, *candidate)) { "selected" } else { "unselected" }, @@ -414,12 +450,17 @@ fn select_workload_impl( Replacement::Rewrite(_) => "rewrite", Replacement::ExactComposition(_) => "exact_composition", }, - "estimated_cost": cost.is_finite().then_some(*cost), - "estimated_cost_status": if cost.is_finite() { "available" } else { "not_reported_by_cost_model" }, + "accuracy_status": accuracy_status, + "guarantee": guarantee, + "runtime_support_status": runtime_support_status, + "decision_reason": decision_reason, + "estimated_cost": candidate_cost.map(|cost| cost.0), + "estimated_cost_status": if candidate_cost.is_some() { "available" } else { "unavailable" }, + "cost_estimate": backend_cost_model.and_then(|model| model.candidate_cost_estimate(candidate)), "selected": chosen.is_some_and(|chosen| std::ptr::eq(chosen, *candidate)), - })).collect::>(); - let rejected = space.target_subdag_candidates().find(|candidates| Rc::ptr_eq(&candidates.target, group.target)) - .into_iter().flat_map(|candidates| &candidates.rejected).map(|candidate| serde_json::json!({ + })}).collect::>(); + let rejected = space.target_subdag_candidates().find(|memo| Rc::ptr_eq(&memo.target, group.target)) + .into_iter().flat_map(|memo| &memo.rejected).map(|candidate| serde_json::json!({ "status": "rejected", "strategy": candidate.strategy, "description": candidate.description, "reason": candidate.error.to_string() })).collect::>(); @@ -452,33 +493,25 @@ fn select_workload_impl( Ok(roots) } -/// Select from Planner's legal candidates with deployment-supplied accuracy -/// models and typed evidence (for example a TopK membership certificate). -pub fn select_summary_with_evidence( +/// Evidence and accuracy hooks feed canonical global selection; enumeration +/// order never authorizes a query plan. +pub fn select_query_with_models( expr: &QueryExpr, cost_model: &dyn CostModel, accuracy_model: &dyn AccuracyModel, - allocator: &dyn AccuracyBudgetAllocator, evidence: &dyn AccuracyEvidenceProvider, ) -> Result, SelectionError> { - let root = Rc::new(expr.clone()); - let strategy = SketchAlgorithmStrategy::new_with_planning_inputs_and_evidence( - cost_model, + let strategies = replacement_strategies(cost_model, evidence, accuracy_model); + let space = asap_aware_mapping::search_workload_with_targets( + vec![(0, Rc::new(expr.clone()), None)], + &strategies, accuracy_model, - allocator, - evidence, ); - let candidate = strategy - .replacements(&TargetSubDAG::new(&root)) - .into_iter() - .next() - .ok_or(SelectionError::NoLegalCandidate)?; - match candidate.replacement { - Replacement::Summary(node) => Ok(node), - Replacement::Rewrite(_) | Replacement::ExactComposition(_) => { - Err(SelectionError::UnexpectedRewrite) - } - } + space + .global_selection(cost_model) + .assemble_selected_dag(&space.roots[0].1) + .map_err(|error| SelectionError::Workload(error.to_string()))? + .ok_or(SelectionError::NoLegalCandidate) } #[cfg(test)] @@ -542,6 +575,63 @@ mod workload_tests { ); } + /// Explain availability from the candidate-cost API, even if a display + /// estimate exists, and never label an unknown guarantee as selected. + #[test] + fn explain_keeps_uncertified_and_uncosted_candidates_explicit() { + struct Uncosted; + impl CostModel for Uncosted { + fn rank_candidates( + &self, + _intent: &AggIntent, + candidates: &[planner_types::post_asap::SketchAlgorithm], + ) -> Vec { + candidates.to_vec() + } + + fn candidate_cost( + &self, + _candidate: &asap_aware_mapping::ReplacementSubDAG, + _target: &TargetSubDAG<'_>, + ) -> Option { + None + } + } + + let accuracy = AccuracyTarget::Epsilon(0.05); + let root = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.9,m[1m]) / quantile_over_time(0.5,m[1m])", + accuracy.clone(), + ) + .unwrap(); + let mut trace = serde_json::Value::Null; + select_workload_impl( + vec![(0, Rc::new(root))], + accuracy, + &Uncosted, + &asap_aware_mapping::NoAccuracyEvidence, + &asap_aware_mapping::DefaultAccuracyModel, + Some(&mut trace), + None, + ) + .unwrap(); + let candidates = trace["groups"] + .as_array() + .unwrap() + .iter() + .flat_map(|group| group["candidates"].as_array().unwrap()); + let mut saw_unknown = false; + for candidate in candidates { + assert_eq!(candidate["estimated_cost_status"], "unavailable"); + assert!(candidate["estimated_cost"].is_null()); + if candidate["accuracy_status"] == "unknown" { + saw_unknown = true; + assert_eq!(candidate["selected"], false); + } + } + assert!(saw_unknown); + } + // JSON must not alias NaN and infinity through its null representation. #[test] fn explain_nonfinite_identity_is_unavailable() { @@ -687,12 +777,10 @@ mod workload_tests { ); } - // HydraGroupingStrategy is registered, but a shared grid is only legal - // with a collision bound to compose: `QueryEvidence` (compiler.rs) reports - // none today, so the strategy correctly offers nothing rather than an - // unbounded guarantee. Pin both halves — the wiring and the missing input. + // Missing shared-grid evidence keeps Hydra candidates visible but + // uncertified. Supplying a certificate makes them eligible for selection. #[test] - fn hydra_candidates_wait_for_shared_grid_evidence() { + fn hydra_candidates_remain_visible_without_shared_grid_evidence() { use asap_aware_mapping::{AccuracyEvidenceProvider, PropagationStats}; use planner_types::post_asap::{CompositionOperator, SketchQuery}; use planner_types::pre_asap::query_expr::Source; @@ -749,11 +837,18 @@ mod workload_tests { evidence, ) .replacements(&target) - .len() }; - assert_eq!(hydra(&asap_aware_mapping::NoAccuracyEvidence), 0); + let unknown = hydra(&asap_aware_mapping::NoAccuracyEvidence); + assert_eq!(unknown.len(), 2); + assert!(unknown + .iter() + .all(|candidate| candidate.has_missing_accuracy_evidence())); // HydraCms over Cms and HydraCountSketch over CountSketch. - assert_eq!(hydra(&MeasuredSharedGrid), 2); + let certified = hydra(&MeasuredSharedGrid); + assert_eq!(certified.len(), 2); + assert!(certified + .iter() + .all(|candidate| !candidate.has_missing_accuracy_evidence())); } // A shared aggregate must keep the sketch plan an unshared one gets: diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 15ce225a..bf6b7318 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -9,8 +9,6 @@ pub use asap_types::query_plan::*; use asap_types::PolicyFingerprint; use planner_types::post_asap::{SummaryExpr, SummaryFamilyType, SummaryNode}; use planner_types::pre_asap::Reduction; -#[deprecated(note = "Use query_plan::residual")] -pub use residual as logical; use std::collections::BTreeMap; #[cfg(test)] use std::collections::BTreeSet; @@ -1178,7 +1176,7 @@ mod tests { planner_types::types::AccuracyTarget::Exact, ) .unwrap(); - let root = crate::planner_selection::select_summary_default(&canonical).unwrap(); + let root = crate::planner_selection::plan_test_query(&canonical).unwrap(); let SummaryExpr::BinaryOp { operator, .. } = &root.expr else { panic!("expected the Planner's average rewrite"); }; diff --git a/control_plane/src/query_plan/residual.rs b/control_plane/src/query_plan/residual.rs index 83fcdda0..7e64c008 100644 --- a/control_plane/src/query_plan/residual.rs +++ b/control_plane/src/query_plan/residual.rs @@ -521,10 +521,15 @@ pub(crate) fn selected_residual_nodes( ) else { continue; }; - let Ok(witness) = crate::planner_selection::select_summary_default(&canonical) else { - continue; + // Match provenance against all exact candidates. Do not make a + // second selection or assume the first enumerated candidate won. + use asap_aware_mapping::{ + Replacement, ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG, }; - if witness.as_ref() == selected { + let root = std::rc::Rc::new(canonical); + let candidates = SketchAlgorithmStrategy::new(&asap_aware_mapping::DefaultCostModel) + .replacements(&TargetSubDAG::new(&root)); + if candidates.iter().any(|candidate| matches!(&candidate.replacement, Replacement::Summary(node) if node.as_ref() == selected)) { let mut lower = Lower { nodes: BTreeMap::new(), seen: BTreeMap::new(), @@ -579,7 +584,7 @@ mod hybrid_tests { planner_types::types::AccuracyTarget::Exact, ) .unwrap(); - let selected = crate::planner_selection::select_summary_default(&canonical).unwrap(); + let selected = crate::planner_selection::plan_test_query(&canonical).unwrap(); let entry = crate::query_plan::compile_bound_composable_mapped( "hybrid".into(), @@ -653,7 +658,7 @@ mod hybrid_tests { planner_types::types::AccuracyTarget::Exact, ) .unwrap(); - let selected = crate::planner_selection::select_summary_default(&canonical).unwrap(); + let selected = crate::planner_selection::plan_test_query(&canonical).unwrap(); assert!( selected_residual_nodes("sum_over_time(m{job=\"worker\"}[5m])", &selected).is_err() ); @@ -783,8 +788,19 @@ mod planner_workload_tests { planner_types::types::AccuracyTarget::Exact, ) .unwrap(); - let selected = crate::planner_selection::select_summary_default(&canonical).unwrap(); - let operator = selected_aggregate_operator(query, &selected).unwrap(); + use asap_aware_mapping::{ + Replacement, ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG, + }; + let root = std::rc::Rc::new(canonical); + let candidates = SketchAlgorithmStrategy::new(&asap_aware_mapping::DefaultCostModel) + .replacements(&TargetSubDAG::new(&root)); + let [candidate] = candidates.as_slice() else { + panic!("expected one exact aggregate candidate") + }; + let Replacement::Summary(selected) = &candidate.replacement else { + panic!("expected exact summary fixture") + }; + let operator = selected_aggregate_operator(query, selected).unwrap(); assert!(matches!( operator, ResidualQueryOperator::Aggregate { @@ -802,13 +818,13 @@ mod planner_workload_tests { planner_types::types::AccuracyTarget::Exact, ) .unwrap(); - let selected = crate::planner_selection::select_summary_default(&canonical).unwrap(); + let selected = crate::planner_selection::plan_test_query(&canonical).unwrap(); let maximum = crate::query_parser::parse_query_expr_canonical( "max(m)", planner_types::types::AccuracyTarget::Exact, ) .unwrap(); - let maximum = crate::planner_selection::select_summary_default(&maximum).unwrap(); + let maximum = crate::planner_selection::plan_test_query(&maximum).unwrap(); let result = selected_residual_nodes("min(m) + max(m)", &selected); if selected == maximum { assert!(result.is_err()); @@ -910,7 +926,7 @@ mod range_max_materialization_tests { planner_types::types::AccuracyTarget::Exact, ) .unwrap(); - let selected = crate::planner_selection::select_summary_default(&original).unwrap(); + let selected = crate::planner_selection::plan_test_query(&original).unwrap(); let key = selected_range_max_materialization(query, &selected) .unwrap() .unwrap(); @@ -930,7 +946,7 @@ mod range_max_materialization_tests { planner_types::types::AccuracyTarget::Exact, ) .unwrap(); - let selected = crate::planner_selection::select_summary_default(&original).unwrap(); + let selected = crate::planner_selection::plan_test_query(&original).unwrap(); assert!( selected_range_max_materialization(query, &selected) .unwrap() @@ -941,7 +957,7 @@ mod range_max_materialization_tests { } } -/// Stable contract identity used by priced physical alternatives, independent of node IDs. +/// Stable contract identity used by priced physical candidates, independent of node IDs. fn materialization_candidate_key( candidate: MaterializationCandidateIdentity, ) -> Result { @@ -1323,7 +1339,7 @@ mod remote_boundary_regressions { planner_types::types::AccuracyTarget::Exact, ) .unwrap(); - let selected = crate::planner_selection::select_summary_default(&parsed).unwrap(); + let selected = crate::planner_selection::plan_test_query(&parsed).unwrap(); assert_eq!( eligible_materialization_keys(query, &selected) .unwrap() @@ -1333,9 +1349,6 @@ mod remote_boundary_regressions { } } -#[deprecated(note = "Use eligible_materialization_keys")] -pub use eligible_materialization_keys as materialization_candidate_keys; - #[cfg(test)] mod tests { use super::*; diff --git a/crates/asap_types/src/query_plan.rs b/crates/asap_types/src/query_plan.rs index 5bdd2605..7595ee93 100644 --- a/crates/asap_types/src/query_plan.rs +++ b/crates/asap_types/src/query_plan.rs @@ -8,9 +8,6 @@ pub mod current_series; pub mod residual; -#[deprecated(note = "Use query_plan::residual")] -pub use residual as logical; - use std::collections::{BTreeMap, BTreeSet}; use planner_types::post_asap::SketchQuery; diff --git a/crates/asap_types/src/query_plan/current_series.rs b/crates/asap_types/src/query_plan/current_series.rs index a4b7480f..4eb69cf1 100644 --- a/crates/asap_types/src/query_plan/current_series.rs +++ b/crates/asap_types/src/query_plan/current_series.rs @@ -1,6 +1,6 @@ //! Maintained current-value populations, shared independently of q and k. use super::{ - logical::{Grouping, LabelMatcher}, + residual::{Grouping, LabelMatcher}, QueryPlanError, }; use serde::{Deserialize, Serialize}; diff --git a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs index 2a451d6a..b5b82264 100644 --- a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs +++ b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs @@ -2,7 +2,7 @@ use super::logical_dag::{PreparedLeaf, PreparedLeaves, Value}; use crate::query_engines::EngineError; use asap_types::query_plan::{ - logical::ResidualQueryOperator, ExternalExactInput, ExternalExactRequest, QueryLanguage, + residual::ResidualQueryOperator, ExternalExactInput, ExternalExactRequest, QueryLanguage, QueryNodeId, QueryPlanEntry, QueryPlanNode, }; use std::collections::{BTreeMap, BTreeSet, HashMap}; @@ -651,7 +651,7 @@ mod tests { #[tokio::test] async fn candidate_exact_is_discovered_and_prepared_behind_candidate_topk_root() { - use asap_types::query_plan::{logical::Grouping, CandidateCompleteness}; + use asap_types::query_plan::{residual::Grouping, CandidateCompleteness}; let mut entry = candidate_entry("sum by (job) (rate(m[5m]))"); entry.nodes.insert( QueryNodeId(2), @@ -888,7 +888,7 @@ mod tests { }; use crate::storage_engines::types::{KeyByLabelValues, Measurement}; use asap_types::query_plan::{ - logical::BinaryOperation, ExactReadout, MaterializationBinding, PhysicalGrouping, + residual::BinaryOperation, ExactReadout, MaterializationBinding, PhysicalGrouping, }; use std::sync::{ atomic::{AtomicUsize, Ordering}, diff --git a/data_plane/src/storage_engines/sketch_db/current_series.rs b/data_plane/src/storage_engines/sketch_db/current_series.rs index 2ba9e584..efd3c012 100644 --- a/data_plane/src/storage_engines/sketch_db/current_series.rs +++ b/data_plane/src/storage_engines/sketch_db/current_series.rs @@ -2,7 +2,7 @@ use crate::drivers::ingest::prometheus_remote_write::CanonicalSample; use asap_types::query_plan::{ current_series::{SeriesPopulation, SeriesReadout}, - logical::{LabelMatch, ResidualQueryOperator}, + residual::{LabelMatch, ResidualQueryOperator}, QueryPlan, QueryPlanNode, }; use std::collections::{BTreeMap, BTreeSet}; diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 51df1d98..55d9b165 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -90,6 +90,111 @@ fn quote_snapshot_for_frontend_test( snapshot } +/// Uncertified candidates must route through the installed exact endpoint unchanged. +async fn assert_uncertified_exact_process(fixture: Value, queries: &[&str]) { + use control_plane::physical::compiler::BackendLocalPlanningInput; + use control_plane::query_plan::QueryPlanNode; + let snapshot: BackendLocalPlanningInput = serde_json::from_value(fixture).unwrap(); + let priced = quote_snapshot_for_test(snapshot); + let plan = priced.clone().compile_promql().unwrap(); + assert!( + plan.precompute_plan.materializations.is_empty(), + "{plan:#?}" + ); + for entry in plan.query_plan.entries.values() { + assert!( + entry.nodes.values().any(|node| matches!( + node, + QueryPlanNode::ExactFallback { .. } | QueryPlanNode::ExternalExact { .. } + )), + "{entry:#?}" + ); + assert!( + !entry + .nodes + .values() + .any(|node| matches!(node, QueryPlanNode::SummaryEstimate { .. })), + "{entry:#?}" + ); + } + let received = Arc::new(Mutex::new(Vec::>::new())); + let requests = received.clone(); + let exact_response = serde_json::json!({ + "status": "success", "data": {"resultType": "vector", "result": [ + {"metric": {"instance": "a"}, "value": [12345, "17"]} + ]} + }); + let expected = exact_response.clone(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let fallback_url = format!("http://{}", listener.local_addr().unwrap()); + let fallback = tokio::spawn(async move { + axum::serve( + listener, + Router::new() + .route("/-/healthy", get(|| async { "healthy" })) + .route( + "/api/v1/query", + get(move |Query(params): Query>| { + let requests = requests.clone(); + let response = exact_response.clone(); + async move { + requests.lock().await.push(params); + Json(response) + } + }), + ), + ) + .await + .unwrap(); + }); + let output = tempfile::tempdir().unwrap(); + let path = output.path().join("planning.json"); + 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")) + .args(["--profile", "asapquery", "--planning-snapshot"]) + .arg(path) + .args([ + "--prometheus-server", + &fallback_url, + "--forward-unsupported-queries", + "--http-port", + &port.to_string(), + "--output-dir", + ]) + .arg(output.path()) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(), + ); + let client = reqwest::Client::new(); + let backend = format!("http://127.0.0.1:{port}"); + wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await; + for query in queries { + let response = client + .get(format!("{backend}/api/v1/query")) + .query(&[("query", *query), ("time", "12345")]) + .send() + .await + .unwrap(); + assert!( + response.status().is_success(), + "{}", + response.text().await.unwrap() + ); + assert_eq!(response.json::().await.unwrap(), expected); + } + let received = received.lock().await; + assert_eq!(received.len(), queries.len()); + for (request, query) in received.iter().zip(queries) { + assert_eq!(request.get("query").unwrap(), query); + assert_eq!(request.get("time").unwrap(), "12345"); + } + fallback.abort(); +} + struct ChildGuard(Child); impl Drop for ChildGuard { @@ -548,11 +653,10 @@ async fn registered_temporal_topk(algorithm: planner_types::post_asap::SketchAlg query.accuracy_target.clone(), algorithm.clone(), ); - query.selected_plan_root = control_plane::planner_selection::select_summary_with_evidence( + query.selected_plan_root = control_plane::planner_selection::select_query_with_models( &expr, &model, &asap_aware_mapping::DefaultAccuracyModel, - &asap_aware_mapping::EqualSplitAllocator, &Evidence, ) .unwrap(); diff --git a/data_plane/tests/backend_process_e2e.rs b/data_plane/tests/backend_process_e2e.rs index cd0f0e86..29a5b178 100644 --- a/data_plane/tests/backend_process_e2e.rs +++ b/data_plane/tests/backend_process_e2e.rs @@ -549,13 +549,13 @@ 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"] + publication["cost_comparison"]["candidates"] .as_array() .unwrap() .len(), 2 ); - assert!(publication["cost_comparison"]["alternatives"][1]["unavailable_reason"].is_string()); + assert!(publication["cost_comparison"]["candidates"][1]["unavailable_reason"].is_string()); let (collector_plan, collector_socket) = collector.await.expect("collector task completed"); assert_eq!( publication["plan_id"], diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs index 8463c9c4..84e7af7e 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -9,8 +9,8 @@ //! query answer //! //! The control plane drives the plan: a PromQL query and an accuracy target -//! go through `BackendLocalPlanningSnapshot::planning_request` → -//! `PhysicalCompiler::compile`, and the resulting materializations are +//! go through `BackendLocalPlanningInput::planning_request` → +//! `PhysicalPlanCompiler::compile`, and the resulting materializations are //! projected into a physical-plan artifact with QueryPlan/SummaryCatalog //! bindings, then staged and activated before ingest. //! diff --git a/data_plane/tests/support/current_series_process.rs b/data_plane/tests/support/current_series_process.rs index be23ae92..b0dabba0 100644 --- a/data_plane/tests/support/current_series_process.rs +++ b/data_plane/tests/support/current_series_process.rs @@ -1,6 +1,8 @@ use super::*; use control_plane::physical::{ - compiler::{BackendLocalPlanningInput, PhysicalCompiler, BACKEND_REVISION, PLANNER_REVISION}, + compiler::{ + BackendLocalPlanningInput, PhysicalPlanCompiler, BACKEND_REVISION, PLANNER_REVISION, + }, workload_cost::{self, WorkloadCostEvidence, WorkloadQuote}, }; @@ -62,11 +64,11 @@ async fn current_series_quantiles_topk_share_and_replace_values() { .clone() .into_physical_compilation_request() .unwrap(); - let candidates = workload_cost::with_exact_alternative(request).unwrap(); + let candidates = workload_cost::enumerate_exact_and_materialized_candidates(request).unwrap(); let quotes = candidates .into_iter() .filter_map(|candidate| { - let plan = PhysicalCompiler + let plan = PhysicalPlanCompiler .compile_promql(candidate.clone(), env.clone()) .ok()?; let warm = candidate.queries.iter().all(|query| { diff --git a/data_plane/tests/support/distinct_planning_process.rs b/data_plane/tests/support/distinct_planning_process.rs index 058ca9d7..72bb3dd7 100644 --- a/data_plane/tests/support/distinct_planning_process.rs +++ b/data_plane/tests/support/distinct_planning_process.rs @@ -1,20 +1,9 @@ use super::*; -use control_plane::physical::compiler::BackendLocalPlanningInput; -/// The production compiler, ingest engine and query DAG preserve distinct populations. +/// Modeled HLL error without a confidence certificate cannot replace exact execution. #[tokio::test] -async fn distinct_range_uses_planner_selected_hll_and_source_labels() { +async fn uncertified_distinct_uses_exact_process() { const QUERY: &str = "distinct_over_time(distinct_values{job=\"api\"}[5s])"; - let fallback_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let fallback_url = format!("http://{}", fallback_listener.local_addr().unwrap()); - let fallback_task = tokio::spawn(async move { - axum::serve( - fallback_listener, - Router::new().route("/-/healthy", get(|| async { "healthy" })), - ) - .await - .unwrap(); - }); let mut fixture: Value = serde_json::from_str(include_str!( "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) @@ -23,147 +12,5 @@ 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 = quote_snapshot_for_test( - serde_json::from_value::(fixture.clone()).unwrap(), - ) - .compile_promql() - .unwrap(); - assert_eq!(plan.precompute_plan.materializations.len(), 1); - assert_eq!( - plan.precompute_plan.materializations[0].aggregation_type, - asap_types::AggregationType::HLL - ); - eprintln!( - "DISTINCT_PLANNED {}", - serde_json::json!({"materializations": plan.precompute_plan.materializations, "query_plan": plan.query_plan, "lifecycle_estimates": plan.lifecycle_estimates}) - ); - let output = tempfile::tempdir().unwrap(); - let path = output.path().join("planning.json"); - 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 { - vm_port = unused_port(); - } - let mut child = ChildGuard( - Command::new(env!("CARGO_BIN_EXE_data_plane")) - .args([ - "--forward-unsupported-queries", - "--prometheus-server", - &fallback_url, - "--profile", - "asapquery", - "--planning-snapshot", - ]) - .arg(&path) - .args(["--http-port", &port.to_string(), "--output-dir"]) - .arg(output.path()) - .args([ - "--victoriametrics-http-port", - &vm_port.to_string(), - "--victoriametrics-url", - &fallback_url, - ]) - .args([ - "--precompute-allowed-lateness-ms", - "0", - "--precompute-flush-interval-ms", - "25", - ]) - .stdout(Stdio::null()) - .stderr(Stdio::inherit()) - .spawn() - .unwrap(), - ); - let client = reqwest::Client::new(); - let backend = format!("http://127.0.0.1:{port}"); - wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await; - // 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 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 { - summary_catalog: compiled.summary_catalog, - collector_plans: compiled.collector_plans, - precompute_plan: compiled.precompute_plan, - transmission_plan: compiled.transmission_plan, - query_plan: compiled.query_plan, - storage_routing: None, - adaptation_evidence: vec![], - }; - eprintln!( - "DISTINCT_INSTALLED {}", - serde_json::to_string(&install).unwrap() - ); - let response = client - .post(format!("{backend}/api/v1/physical-plan")) - .json(&install) - .send() - .await - .unwrap(); - assert!( - response.status().is_success(), - "{}", - response.text().await.unwrap() - ); - let response = client - .post(format!("{backend}/api/v1/physical-plan/activate")) - .json(&identity) - .send() - .await - .unwrap(); - assert!( - response.status().is_success(), - "{}", - response.text().await.unwrap() - ); - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis() as i64; - let base = now - now.rem_euclid(5000) - 20000; - let mut series = Vec::new(); - for (instance, distinct, job) in [("a", 5, "api"), ("b", 13, "api"), ("excluded", 23, "other")] - { - let mut samples: Vec<_> = (0..100) - .map(|i| (base + 1 + i, (i % distinct) as f64)) - .collect(); - samples.push((base + 15001, 1000.0)); - series.push(series_with_labels( - "distinct_values", - &[("instance", instance), ("job", job)], - &samples, - )); - } - assert_eq!( - remote_write(&client, &backend, &WriteRequest { timeseries: series }).await, - 204 - ); - drain_precompute(&client, &backend).await; - let result = wait_for_warm_instant( - &client, - &format!("http://127.0.0.1:{vm_port}"), - QUERY, - (base + 5000) as f64 / 1000.0, - &output.path().join("query_engine.log"), - ) - .await; - let rows = result["data"]["result"].as_array().unwrap(); - assert_eq!(rows.len(), 2, "{result}"); - for (instance, exact) in [("a", 5.0), ("b", 13.0)] { - let row = rows - .iter() - .find(|row| row["metric"]["instance"] == instance) - .unwrap(); - let estimate = row["value"][1].as_str().unwrap().parse::().unwrap(); - assert!((estimate - exact).abs() / exact <= 0.05, "{result}"); - } - eprintln!("DISTINCT_WARM {result}"); - fallback_task.abort(); + assert_uncertified_exact_process(fixture, &[QUERY]).await; } diff --git a/data_plane/tests/support/erp_planning_process.rs b/data_plane/tests/support/erp_planning_process.rs index 59b32c89..1f1bb9db 100644 --- a/data_plane/tests/support/erp_planning_process.rs +++ b/data_plane/tests/support/erp_planning_process.rs @@ -1,5 +1,5 @@ use super::*; -use control_plane::physical::{compiler::BackendLocalPlanningInput, erp::ErpShapeObserver}; +use control_plane::physical::erp::ErpShapeObserver; fn measured_profiles(raw: &[f64]) -> Value { let mut records = Vec::new(); @@ -39,17 +39,7 @@ fn measured_profiles(raw: &[f64]) -> Value { } #[tokio::test] -async fn observed_shape_selects_installed_parameters_and_executes_remote_write() { - let fallback_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let fallback_url = format!("http://{}", fallback_listener.local_addr().unwrap()); - let fallback_task = tokio::spawn(async move { - axum::serve( - fallback_listener, - Router::new().route("/-/healthy", get(|| async { "healthy" })), - ) - .await - .unwrap(); - }); +async fn measured_error_without_confidence_uses_exact_process() { const QUERY: &str = "quantile_over_time(0.9, erp_latency[5s])"; let training: Vec = (1..=16) .flat_map(|value| std::iter::repeat_n(value as f64, 512 / value)) @@ -64,7 +54,6 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write() } let observation = observer.snapshot().unwrap(); let artifact = measured_profiles(&training); - let mut chosen = Vec::new(); for only_large in [false, true] { let mut evidence = artifact.clone(); if only_large { @@ -102,163 +91,6 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write() ), control_plane::physical::erp::ErpParameterDecision::Empirical { .. } )); - let snapshot: BackendLocalPlanningInput = serde_json::from_value(fixture.clone()).unwrap(); - let plan = quote_snapshot_for_test(snapshot).compile_promql().unwrap(); - assert_eq!( - plan.precompute_plan.materializations.len(), - 1, - "plan={plan:#?}; observation={observation:#?}; evidence={artifact}" - ); - let expected_k = if only_large { 128 } else { 32 }; - assert_eq!( - plan.precompute_plan.materializations[0].parameters["k"], - expected_k - ); - chosen.push(plan.precompute_plan.materializations[0].policy_fingerprint()); - eprintln!( - "ERP_PLANNED {}", - serde_json::json!({ - "query": QUERY, "available_profiles": policy.artifact.records, - "parameter_decision": format!("{:?}", policy.select(planner_types::post_asap::SketchAlgorithm::Kll, 0.2, planner_types::post_asap::SketchParams::Kll { k: 128 })), - "lifecycle_estimates": plan.lifecycle_estimates, - "observation": policy.observed_shape, - "selected_parameters": plan.precompute_plan.materializations[0].parameters, - "materialization": chosen.last(), - "partitioning": plan.precompute_plan.materializations[0].partitioning, - "query_plan": plan.query_plan, - }) - ); - let output = tempfile::tempdir().unwrap(); - let path = output.path().join("planning.json"); - 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")) - .args([ - "--forward-unsupported-queries", - "--prometheus-server", - &fallback_url, - "--profile", - "asapquery", - "--planning-snapshot", - ]) - .arg(&path) - .args(["--http-port", &port.to_string(), "--output-dir"]) - .arg(output.path()) - .args([ - "--precompute-allowed-lateness-ms", - "0", - "--precompute-flush-interval-ms", - "25", - ]) - .stdout(Stdio::null()) - .stderr(Stdio::inherit()) - .spawn() - .unwrap(), - ); - let client = reqwest::Client::new(); - let backend = format!("http://127.0.0.1:{port}"); - wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await; - let config: Value = client - .get(format!("{backend}/api/v1/physical-plan/status")) - .send() - .await - .unwrap() - .json() - .await - .unwrap(); - let config_text = serde_json::to_string(&config).unwrap(); - assert!( - config_text.contains(&chosen.last().unwrap().0.to_string()), - "installed ERP identity missing: {config}" - ); - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis() as i64; - let base = now - now.rem_euclid(5000) - 20000; - let samples: Vec<_> = raw - .iter() - .enumerate() - .map(|(i, value)| (base + 1 + i as i64, *value)) - .collect(); - assert_eq!( - remote_write( - &client, - &backend, - &WriteRequest { - timeseries: vec![ - series_with_labels("erp_latency", &[("instance", "a")], &samples), - series_with_labels( - "erp_latency", - &[("instance", "b")], - &samples - .iter() - .map(|(t, v)| (*t, *v + 1000.0)) - .collect::>() - ), - ] - } - ) - .await, - 204 - ); - assert_eq!( - remote_write( - &client, - &backend, - &WriteRequest { - timeseries: vec![ - series_with_labels( - "erp_latency", - &[("instance", "a")], - &[(base + 15001, 1.0)] - ), - series_with_labels( - "erp_latency", - &[("instance", "b")], - &[(base + 15001, 1001.0)] - ), - ] - } - ) - .await, - 204 - ); - drain_precompute(&client, &backend).await; - let result = wait_for_warm_instant( - &client, - &backend, - QUERY, - (base + 5000) as f64 / 1000.0, - &output.path().join("query_engine.log"), - ) - .await; - let rows = result["data"]["result"].as_array().unwrap(); - assert_eq!( - rows.len(), - 2, - "per-series KLL states must not pool: {result}" - ); - eprintln!( - "ERP_WARM {}", - serde_json::json!({"materialization": chosen.last(), "result": result}) - ); - for (instance, offset) in [("a", 0.0), ("b", 1000.0)] { - let row = rows - .iter() - .find(|row| row["metric"]["instance"] == instance) - .expect("source labels retained"); - let estimate = row["value"][1].as_str().unwrap().parse::().unwrap() - offset; - let lower = raw.iter().filter(|v| **v < estimate).count() as f64 / raw.len() as f64; - let upper = raw.iter().filter(|v| **v <= estimate).count() as f64 / raw.len() as f64; - assert!((lower - 0.9).max(0.9 - upper).max(0.0) <= 0.2, "{result}"); - } + assert_uncertified_exact_process(fixture, &[QUERY]).await; } - fallback_task.abort(); - assert_ne!( - chosen[0], chosen[1], - "changed evidence must change installed state identity" - ); } diff --git a/data_plane/tests/support/issue_701_702_process.rs b/data_plane/tests/support/issue_701_702_process.rs index d1fe869e..cdd25157 100644 --- a/data_plane/tests/support/issue_701_702_process.rs +++ b/data_plane/tests/support/issue_701_702_process.rs @@ -1,7 +1,9 @@ //! Issue workloads execute their selected Planner DAG on the production HTTP path. use super::*; use control_plane::physical::{ - compiler::{BackendLocalPlanningInput, PhysicalCompiler, BACKEND_REVISION, PLANNER_REVISION}, + compiler::{ + BackendLocalPlanningInput, PhysicalPlanCompiler, BACKEND_REVISION, PLANNER_REVISION, + }, workload_cost::{self, WorkloadCostEvidence, WorkloadQuote}, }; @@ -134,8 +136,8 @@ async fn run_warm_workload(queries: Vec<(String, u64, u64)>) { .clone() .into_physical_compilation_request() .unwrap(); - let candidates = workload_cost::with_exact_alternative(request).unwrap(); - let fully_warm = |plan: &control_plane::physical::compiler::PhysicalPlan| { + let candidates = workload_cost::enumerate_exact_and_materialized_candidates(request).unwrap(); + let fully_warm = |plan: &control_plane::physical::compiler::CompiledPhysicalPlan| { plan.query_plan.entries.values().all(|entry| entry.nodes.values().all(|node| !matches!(node, control_plane::query_plan::QueryPlanNode::ExactFallback { .. } | control_plane::query_plan::QueryPlanNode::ExternalExact { .. } @@ -150,14 +152,14 @@ async fn run_warm_workload(queries: Vec<(String, u64, u64)>) { let quotes = candidates .into_iter() .filter_map(|candidate| { - let plan = match PhysicalCompiler.compile_promql(candidate.clone(), environment.clone()) - { - Ok(plan) => plan, - Err(error) => { - errors.push(error.to_string()); - return None; - } - }; + let plan = + match PhysicalPlanCompiler.compile_promql(candidate.clone(), environment.clone()) { + Ok(plan) => plan, + Err(error) => { + errors.push(error.to_string()); + return None; + } + }; let warm = fully_warm(&plan); found |= warm; @@ -340,10 +342,11 @@ fn issue_701_702_uncertified_ratios_require_exact_fallback() { fixture["data_workload"]["data_ingestion_interval"]["value"] = 1000.into(); let snapshot: BackendLocalPlanningInput = serde_json::from_value(fixture).unwrap(); let (request, environment) = snapshot.into_physical_compilation_request().unwrap(); - let candidates = workload_cost::with_exact_alternative(request).unwrap(); + let candidates = + workload_cost::enumerate_exact_and_materialized_candidates(request).unwrap(); assert!(!candidates.is_empty()); for candidate in candidates { - let plan = PhysicalCompiler + let plan = PhysicalPlanCompiler .compile_promql(candidate, environment.clone()) .unwrap(); assert!(plan.precompute_plan.materializations.is_empty(), "{query}"); diff --git a/data_plane/tests/support/physical_fixture.rs b/data_plane/tests/support/physical_fixture.rs index 9eb34e64..40329614 100644 --- a/data_plane/tests/support/physical_fixture.rs +++ b/data_plane/tests/support/physical_fixture.rs @@ -46,7 +46,7 @@ pub fn artifact_from_materializations( let mut precompute = PrecomputePlan::build(envelope.clone(), configs, &["fixture".into()]).unwrap(); precompute.summary_catalog = Some(catalog.reference().unwrap()); - let mut transmission = control_plane::physical::compiler::compile_transmission_plan( + let mut transmission = control_plane::physical::compiler::build_transmission_plan( envelope, &precompute, &BTreeMap::new(), diff --git a/data_plane/tests/support/univmon_erp_process.rs b/data_plane/tests/support/univmon_erp_process.rs index e21cebe6..28bb7b01 100644 --- a/data_plane/tests/support/univmon_erp_process.rs +++ b/data_plane/tests/support/univmon_erp_process.rs @@ -1,5 +1,5 @@ use super::*; -use control_plane::physical::{compiler::BackendLocalPlanningInput, erp::ErpShapeObserver}; +use control_plane::physical::erp::ErpShapeObserver; use data_plane::precompute_engine::operators::univmon_accumulator::UnivMonAccumulator; use data_plane::storage_engines::types::{AggregateCore, SerializableToSink}; @@ -83,11 +83,10 @@ fn measured_artifact() -> Value { } #[tokio::test] -async fn measured_readout_evidence_selects_and_executes_univmon() { +async fn measured_univmon_without_confidence_uses_exact_process() { let artifact = measured_artifact(); eprintln!("UNIVMON_MEASURED {artifact}"); let raw = values(100_000); - let exact = truth(&raw); let mut observer = ErpShapeObserver::new(128).unwrap(); for (i, value) in raw.iter().enumerate() { observer.observe(&value.to_string(), i / 100).unwrap(); @@ -126,61 +125,10 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { "minimum_confidence": 0.7, "minimum_confidence_margin": 0.05}, "runtime": {"allowed_algorithms": ["Hll", "Kll", "UnivMon"], "max_memory_bytes": null} }); - let snapshot: BackendLocalPlanningInput = serde_json::from_value(fixture.clone()).unwrap(); - let plan = quote_snapshot_for_test(snapshot).compile_promql().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}) - ); - assert!( - plan.precompute_plan - .materializations - .iter() - .any(|m| m.aggregation_type == asap_types::AggregationType::UnivMon), - "{plan:#?}" - ); - // All three readouts can use one state when the selected parameters and - // population agree. Each still needs its own calibration evidence. - let mut shared_fixture = fixture.clone(); - shared_fixture["implementation"]["erp"]["runtime"]["allowed_algorithms"] = - serde_json::json!(["UnivMon"]); - let records = shared_fixture["implementation"]["erp"]["artifact"]["records"] - .as_array_mut() - .unwrap(); - records.remove(0); - let shared = quote_snapshot_for_test( - serde_json::from_value::(shared_fixture.clone()).unwrap(), - ) - .compile_promql() - .unwrap(); - assert_eq!( - shared.precompute_plan.materializations.len(), - 1, - "distinct, L2 and entropy share one frequency population: {shared:#?}" - ); - assert_eq!( - shared.precompute_plan.materializations[0].aggregation_type, - asap_types::AggregationType::UnivMon - ); - for query in queries { - let entry = shared - .query_plan - .entries - .values() - .find(|e| e.canonical_query == query) - .unwrap(); - assert!( - !entry.nodes.values().any(|n| matches!( - n, - control_plane::query_plan::QueryPlanNode::ExactFallback { .. } - | control_plane::query_plan::QueryPlanNode::ExternalExact { .. } - )), - "{entry:#?}" - ); - } - // Removing only entropy evidence must leave the L2 path executable. - let mut missing_entropy = fixture.clone(); - for row in missing_entropy["implementation"]["erp"]["artifact"]["records"] + // Measured maxima across ten populations are not a failure-probability proof. + assert_uncertified_exact_process(fixture.clone(), &queries).await; + // Removing one readout's measurements cannot authorize the other readouts. + for row in fixture["implementation"]["erp"]["artifact"]["records"] .as_array_mut() .unwrap() { @@ -189,226 +137,5 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { .unwrap() .remove("max_frequency_entropy_absolute_bits_error"); } - let missing = quote_snapshot_for_test( - serde_json::from_value::(missing_entropy).unwrap(), - ) - .compile_promql() - .unwrap(); - use control_plane::query_plan::{QueryPlanNode, QueryReadout}; - assert!(missing - .query_plan - .entries - .values() - .flat_map(|e| e.nodes.values()) - .any(|node| matches!( - node, - QueryPlanNode::SummaryEstimate { - query: QueryReadout::FrequencyL2, - .. - } - ))); - let entropy = missing - .query_plan - .entries - .values() - .find(|e| e.canonical_query.starts_with("entropy_over_time")) - .unwrap(); - assert!( - entropy.nodes.values().any(|node| matches!( - node, - QueryPlanNode::ExactFallback { .. } | QueryPlanNode::ExternalExact { .. } - )), - "{entropy:#?}" - ); - assert!(!entropy.nodes.values().any(|node| matches!( - node, - QueryPlanNode::SummaryEstimate { - query: QueryReadout::FrequencyEntropy, - .. - } - ))); - let plan = shared; - let fixture = shared_fixture; - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let fallback_url = format!("http://{}", listener.local_addr().unwrap()); - let fallback = tokio::spawn(async move { - axum::serve( - listener, - Router::new().route("/-/healthy", get(|| async { "healthy" })), - ) - .await - .unwrap(); - }); - let runtime_samples = control_plane::runtime_samples::RuntimeSamplesStore::new(8); - let runtime_port = unused_port(); - let runtime_endpoint = format!("http://127.0.0.1:{runtime_port}"); - let runtime_service = - control_plane::runtime_samples::RuntimeSamplesService::new(runtime_samples.clone()) - .into_server(); - let runtime_task = tokio::spawn(async move { - tonic::transport::Server::builder() - .add_service(runtime_service) - .serve(([127, 0, 0, 1], runtime_port).into()) - .await - .unwrap(); - }); - let output = tempfile::tempdir().unwrap(); - let path = output.path().join("planning.json"); - 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")) - .args(["--erp-runtime-samples-endpoint", &runtime_endpoint]) - .args(["--profile", "asapquery", "--planning-snapshot"]) - .arg(&path) - .args([ - "--prometheus-server", - &fallback_url, - "--forward-unsupported-queries", - "--http-port", - &port.to_string(), - "--output-dir", - ]) - .arg(output.path()) - .args([ - "--precompute-allowed-lateness-ms", - "0", - "--precompute-flush-interval-ms", - "25", - ]) - .stdout(Stdio::null()) - .stderr(Stdio::inherit()) - .spawn() - .unwrap(), - ); - let client = reqwest::Client::new(); - let backend = format!("http://127.0.0.1:{port}"); - wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await; - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_millis() as i64; - let base = now - now.rem_euclid(5000) - 20_000; - let mut samples: Vec<_> = raw - .iter() - .enumerate() - .map(|(i, v)| (base + 1 + i as i64, *v)) - .collect(); - // Bracket the calibrated population; these boundary samples are outside it. - samples.insert(0, (base, 0.0)); - assert_eq!( - remote_write( - &client, - &backend, - &WriteRequest { - timeseries: vec![series("erp_frequency", &samples)] - } - ) - .await, - 204 - ); - assert_eq!( - remote_write( - &client, - &backend, - &WriteRequest { - timeseries: vec![series("erp_frequency", &[(base + 15001, 0.0)])] - } - ) - .await, - 204 - ); - drain_precompute(&client, &backend).await; - let keys = runtime_samples.keys(); - assert!( - !keys.is_empty(), - "real worker inputs must reach RuntimeSamples after finite drain" - ); - for key in keys { - let record = runtime_samples.latest(&key).unwrap(); - let observed: asap_types::erp_observation::ErpPopulationObservations< - asap_types::erp_observation::EmpiricalFrequencyObservation, - > = serde_json::from_value(record.payload["erp_population_observations"].clone()).unwrap(); - assert!(observed.invalid_reason.is_none(), "{observed:?}"); - assert!(!observed.populations.is_empty()); - assert_eq!(observed.window_end_ms - observed.window_start_ms, 5000); - for population in &observed.populations { - assert_eq!(population.shape.event_count(), Some(raw.len() as u64)); - assert_eq!(population.shape.sorted_counts.len(), 128); - } - assert!(plan - .summary_catalog - .materializations - .contains_key(&observed.summary_definition_id)); - assert_eq!( - observed.catalog_generation, - plan.summary_catalog.reference().unwrap() - ); - if key.sketch == "univmon" { - let mut live_snapshot: BackendLocalPlanningInput = - serde_json::from_value(fixture.clone()).unwrap(); - let policy = live_snapshot.physical_inputs.erp.as_mut().unwrap(); - policy.observed_shape_source = - Some(control_plane::physical::erp::ErpObservedShapeSource { - source: key.source.clone(), - sketch: key.sketch.clone(), - implementation: key.impl_name.clone(), - population_scope: Some( - control_plane::physical::erp::ErpPopulationObservationScope { - catalog_generation: observed.catalog_generation.clone(), - summary_definition_id: observed.summary_definition_id, - input_semantics: observed.input_semantics, - freshness: asap_types::erp_observation::ErpObservationFreshness { - max_age_ms: 60_000, - max_future_skew_ms: 1000, - }, - }, - ), - }); - policy.hydrate_observed_shape(&runtime_samples).unwrap(); - policy.resolve_population_data_descriptor(Some(&plan.summary_catalog)); - assert!(policy - .observed_populations - .as_ref() - .unwrap() - .invalid_reason - .is_none()); - let replanned = quote_snapshot_for_test(live_snapshot) - .compile_promql() - .unwrap(); - assert!( - replanned - .precompute_plan - .materializations - .iter() - .any(|m| m.aggregation_type == asap_types::AggregationType::UnivMon), - "actual producer evidence should reach normal Planner selection" - ); - } - } - runtime_task.abort(); - - for (i, query) in queries.iter().enumerate() { - let result = wait_for_warm_instant( - &client, - &backend, - query, - (base + 5000) as f64 / 1000.0, - &output.path().join("query_engine.log"), - ) - .await; - let estimate = first_value(&result, "value").unwrap(); - let error = (estimate - exact[i]).abs() / if i == 2 { 1.0 } else { exact[i] }; - assert!( - error <= 0.2, - "{query}: {result}, truth={}, error={error}", - exact[i] - ); - eprintln!( - "UNIVMON_WARM {}", - serde_json::json!({"query": query, "result": result, "truth": exact[i], "measured_error": error, "units": if i == 2 { "absolute_bits" } else { "relative" }}) - ); - } - fallback.abort(); + assert_uncertified_exact_process(fixture, &queries).await; }