diff --git a/Cargo.lock b/Cargo.lock index 96ae6df2..7912b94a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -373,7 +373,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" dependencies = [ "asap-types", "serde", @@ -384,7 +384,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" dependencies = [ "asap-types", "promql-parser", @@ -393,7 +393,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -415,12 +415,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=3be523fa0f06a905188e42cbe482d06aa843ba5d#3be523fa0f06a905188e42cbe482d06aa843ba5d" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" dependencies = [ "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 3f475e27..11d2724e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,12 +28,12 @@ version = "0.1.0" asap_sketchlib = { path = "../asap_sketchlib" } [workspace.dependencies] -# Keep Planner frontends, selection, and IR on the same immutable main revision. +# 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 = "3be523fa0f06a905188e42cbe482d06aa843ba5d" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" } -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" } -asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "3be523fa0f06a905188e42cbe482d06aa843ba5d" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" } +asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" } # Shared external deps (used by 2+ crates) serde = { version = "1.0", features = ["derive"] } diff --git a/README.md b/README.md index 57181620..102e1783 100644 --- a/README.md +++ b/README.md @@ -264,7 +264,7 @@ docker version cargo +1.98.0 fetch --locked cargo +1.98.0 build --locked -p control_plane -p data_plane cargo +1.98.0 build --locked -p control_plane \ - --example calibration_candidates --example inspect_physical_dag \ + --example calibration_candidates --example compile_workload_artifact \ --example compile_clickhouse_workload target/debug/data_plane --help mkdir -p target/readme-evidence @@ -278,7 +278,7 @@ mkdir -p target/readme-evidence The executable is `target/debug/data_plane`, or `target/release/data_plane` if you build with `--release`. Access to the pinned Git dependencies is required. -### 3. Export candidates and inspect the ordinary selected plan +### 3. Export candidates and select a deployable plan by complete cost ```bash target/debug/examples/calibration_candidates \ @@ -287,10 +287,10 @@ target/debug/examples/calibration_candidates \ jq '.candidates[] | {candidate_index, unavailable_reason}' \ target/readme-evidence/candidates.json -target/debug/examples/inspect_physical_dag \ - docs/examples/asapquery-compatibility-demo-snapshot.json \ +target/debug/examples/compile_workload_artifact \ + "$ASAPQUERY_PLANNING_SNAPSHOT" \ > target/readme-evidence/selected.json -jq '.purpose' target/readme-evidence/selected.json +jq '.cost_comparison' target/readme-evidence/selected.json jq '.install_request.summary_catalog' target/readme-evidence/selected.json jq '.install_request.precompute_plan | {materializations, executable_dags}' \ target/readme-evidence/selected.json @@ -299,13 +299,13 @@ jq '.install_request.precompute_plan.schemas[] | {materialization, schema_id}' \ target/readme-evidence/selected.json ``` -Expect `inspection_only`, catalog/plan objects and explicit candidate rejection -reasons where unsupported. One verified demo export contained five candidate -entries (one installable), five selected materializations and six query entries; -these are inspection evidence, not a permanent optimizer-count contract. -Materialization IDs are definitions, not physical SIDs. Demo costs are not -measurements. The [E2E walkthrough](docs/evaluation/e2e-physical-dag.md) explains -ERP evidence and the version-2 measured-cost workflow. +Candidate discovery accepts the checked-in unquoted templates. Deployment and +selected-plan inspection require `ASAPQUERY_PLANNING_SNAPSHOT` to point to a +snapshot with complete, valid workload cost evidence. Prepare that input using +the [cost evidence workflow](docs/examples/workload-cost-evidence.md). +There is one snapshot compiler: it compares complete executable alternatives, +including exact fallback. Materialization IDs are definitions, not physical SIDs. +For `--metricsql`, collect quotes for the MetricsQL frontend. ## Prometheus runbook @@ -342,7 +342,7 @@ done curl -fsS http://127.0.0.1:19090/-/healthy target/debug/data_plane --profile asapquery \ - --planning-snapshot docs/examples/asapquery-compatibility-demo-snapshot.json \ + --planning-snapshot "$ASAPQUERY_PLANNING_SNAPSHOT" \ --prometheus-server http://127.0.0.1:19090 \ --forward-unsupported-queries --http-port 19091 \ --output-dir target/readme-evidence/prometheus/runtime \ @@ -409,6 +409,7 @@ kill "$(cat target/readme-evidence/prometheus/backend.pid)" docker rm -f asap-readme-prometheus asap-readme-pushgateway cargo +1.98.0 test --locked -p data_plane --test asapquery_compatibility_process_e2e \ collector_free_profile_serves_complete_matrix_and_falls_back_exactly -- --exact +export ASAPQUERY_PLANNING_SNAPSHOT=/absolute/path/priced-snapshot.json ./scripts/e2e.sh asapquery-demo ``` @@ -467,8 +468,8 @@ kill "$(cat target/readme-evidence/victoriametrics/backend.pid)" To inspect supported MetricsQL planning independently: ```bash -target/debug/examples/inspect_physical_dag \ - docs/examples/asapquery-compatibility-demo-snapshot.json --metricsql \ +target/debug/examples/compile_workload_artifact \ + "$ASAPQUERY_PLANNING_SNAPSHOT" --metricsql \ > target/readme-evidence/victoriametrics/selected.json jq '.install_request.query_plan.entries' \ target/readme-evidence/victoriametrics/selected.json diff --git a/control_plane/docs/candidate-physical-explain.md b/control_plane/docs/candidate-physical-explain.md index 993ce39f..fded1a0a 100644 --- a/control_plane/docs/candidate-physical-explain.md +++ b/control_plane/docs/candidate-physical-explain.md @@ -18,7 +18,7 @@ Physical identity combines the logical/mask alternative with existing materializ The read-only `/api/v1/physical-plan/cost-manifests` and MetricsQL equivalent retain their default manifest-array response. Add `"explain": true` to the existing request to receive `{ "manifests": [...], "alternatives": [...], "logical_selection": [...] }`. Failed alternatives remain alongside usable manifests. When none can bind or be completely priced, the error retains an `all_infeasible` report and every accumulated alternative rather than only a generic message. -Snapshot compilation exposes the same logical trace on `PhysicalPlan`; `inspect_physical_dag` prints it outside the install request. Compile-and-publish returns the trace without adding it to executable wire DTOs. SQL's existing selection trace gains the same semantic candidate/root identities. +Snapshot compilation exposes the same logical trace on `PhysicalPlan`; `compile_workload_artifact` prints it outside the install request. Compile-and-publish returns the trace without adding it to executable wire DTOs. SQL's existing selection trace gains the same semantic candidate/root identities. These are bounded explanations: they cover the actual Planner search and the existing physical materialization/exact inventory, not every possible placement or resource-constrained cluster assignment. Missing numeric measurements remain missing. The next provider integration must occur before logical commitment and reuse Planner's provider/resource contracts. diff --git a/control_plane/examples/compile_workload_artifact.rs b/control_plane/examples/compile_workload_artifact.rs index b9c3c348..fa6ceed5 100644 --- a/control_plane/examples/compile_workload_artifact.rs +++ b/control_plane/examples/compile_workload_artifact.rs @@ -3,17 +3,21 @@ use control_plane::physical::compiler::BackendLocalPlanningSnapshot; use serde_json::json; fn main() -> Result<(), Box> { - let path = std::env::args() - .nth(1) + let mut args = std::env::args().skip(1); + let path = args + .next() .ok_or("usage: compile_workload_artifact SNAPSHOT.json [--metricsql]")?; - let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?; - if snapshot.snapshot_version != 2 { - return Err( - "execution evaluation requires version 2 complete workload cost evidence".into(), - ); + let metricsql = match args.next().as_deref() { + None => false, + Some("--metricsql") => true, + Some(_) => return Err("expected optional --metricsql".into()), + }; + if args.next().is_some() { + return Err("unexpected arguments".into()); } + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?; let start = std::time::Instant::now(); - let plan = if std::env::args().skip(2).any(|arg| arg == "--metricsql") { + let plan = if metricsql { snapshot.compile_metricsql()? } else { snapshot.compile()? @@ -29,6 +33,9 @@ fn main() -> Result<(), Box> { "planning_elapsed_ns": elapsed, "envelope": plan.envelope, "cost_comparison": comparison, + "logical_selection": plan.logical_selection, + "backend_revision": control_plane::physical::compiler::BACKEND_REVISION, + "planner_revision": control_plane::physical::compiler::PLANNER_REVISION, "lifecycle_estimates": plan.lifecycle_estimates, "install_request": { "summary_catalog": plan.summary_catalog, diff --git a/control_plane/examples/inspect_physical_dag.rs b/control_plane/examples/inspect_physical_dag.rs deleted file mode 100644 index b9d812ff..00000000 --- a/control_plane/examples/inspect_physical_dag.rs +++ /dev/null @@ -1,49 +0,0 @@ -//! Inspect an ordinary planning snapshot without treating demo costs as measurements. -use control_plane::physical::compiler::BackendLocalPlanningSnapshot; -use serde_json::json; - -fn main() -> Result<(), Box> { - let mut args = std::env::args().skip(1); - let path = args - .next() - .ok_or("usage: inspect_physical_dag SNAPSHOT.json [--metricsql]")?; - let metricsql = match args.next().as_deref() { - None => false, - Some("--metricsql") => true, - Some(_) => return Err("expected optional --metricsql".into()), - }; - if args.next().is_some() { - return Err("unexpected arguments".into()); - } - let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?; - let snapshot_version = snapshot.snapshot_version; - let erp_supplied = snapshot.implementation.erp.is_some(); - let plan = if metricsql { - snapshot.compile_metricsql()? - } else { - snapshot.compile()? - }; - println!( - "{}", - serde_json::to_string_pretty(&json!({ - "purpose": "inspection_only", - "snapshot_version": snapshot_version, - "erp_input_supplied": erp_supplied, - "backend_revision": control_plane::physical::compiler::BACKEND_REVISION, - "planner_revision": control_plane::physical::compiler::PLANNER_REVISION, - "cost_comparison": plan.cost_comparison, - "logical_selection": plan.logical_selection, - "lifecycle_estimates": plan.lifecycle_estimates, - "install_request": { - "summary_catalog": plan.summary_catalog, - "collector_plans": plan.collector_plans, - "precompute_plan": plan.precompute_plan, - "transmission_plan": plan.transmission_plan, - "query_plan": plan.query_plan, - "storage_routing": null, - "adaptation_evidence": [] - } - }))? - ); - Ok(()) -} diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index fb60d3cc..75841913 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -601,7 +601,11 @@ mod tests { "../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - let publication = snapshot.compile().unwrap().publication().unwrap(); + let publication = crate::physical::compiler::tests::quoted_snapshot(snapshot, false) + .compile() + .unwrap() + .publication() + .unwrap(); let hits: StdArc>> = StdArc::new(Mutex::new(Vec::new())); let route_hits = hits.clone(); let app = Router::new().route( diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 88570956..d3664686 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -971,6 +971,7 @@ fn compile_physical_plan_request( }; let planning_request = physical::compiler::PlanningRequest { + synthesized_window_queries: Default::default(), logical_selection, query_workload: None, queries, diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index b57f00e8..6818c515 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -148,6 +148,8 @@ pub struct PlanningRequest { /// Original dashboard demand, in the same order as queries. None is legacy input. pub query_workload: Option, pub queries: Vec, + /// Compiler-owned quotes eligible for joint pane repricing; never inferred from model labels. + pub synthesized_window_queries: BTreeSet, pub evidence: HashMap, /// Fresh measured costs for Planner exact/summary composition sites, /// scoped to query IDs just like accuracy evidence. @@ -198,7 +200,8 @@ pub enum PhysicalDeploymentTarget { BackendLocalRemoteWrite, } -/// Versioned startup input for the Collector-free compatibility profile. +/// Startup and candidate-discovery input for backend-local planning. +/// Version 2 is the sole supported schema; deployment always requires quotes. /// Query/data semantics use ASAPPlanner's canonical workload types directly; /// this wrapper adds only backend-owned implementation evidence and lifecycle /// identity required to choose a concrete physical realization. @@ -206,6 +209,7 @@ pub enum PhysicalDeploymentTarget { #[serde(deny_unknown_fields)] pub struct BackendLocalPlanningSnapshot { pub snapshot_version: u32, + /// May be absent during candidate discovery, never during deployment. #[serde(default, skip_serializing_if = "Option::is_none")] pub workload_cost_evidence: Option, pub query_workload: QueryWorkload, @@ -480,34 +484,17 @@ impl BackendLocalPlanningSnapshot { } fn compile_frontend(self, metricsql: bool) -> Result { - let evidence = self.workload_cost_evidence.clone(); - if self.snapshot_version == 2 && evidence.is_none() { - return Err(CompileError::Snapshot( - "version 2 requires complete workload cost evidence".into(), - )); - } + let evidence = self.workload_cost_evidence.clone().ok_or_else(|| { + CompileError::Snapshot( + "deployment requires complete workload cost evidence; export candidates and price them before compiling".into(), + ) + })?; let (request, environment) = self.planning_request()?; - match evidence { - Some(evidence) => { - let candidates = super::workload_cost::with_exact_alternative(request)?; - if metricsql { - super::workload_cost::select_metricsql(candidates, environment, &evidence) - } else { - super::workload_cost::select(candidates, environment, &evidence) - } - } - None => { - // Unquoted v1 startup snapshots keep the established summary/native - // compatibility policy. Local residual candidates are enumerated by - // planning_request and admitted through measured workload selection. - let mut request = request; - request.hybrid_execution = false; - if metricsql { - PhysicalCompiler.compile_metricsql(request, environment) - } else { - PhysicalCompiler.compile(request, environment) - } - } + let candidates = super::workload_cost::with_exact_alternative(request)?; + if metricsql { + super::workload_cost::select_metricsql(candidates, environment, &evidence) + } else { + super::workload_cost::select(candidates, environment, &evidence) } } @@ -515,9 +502,9 @@ impl BackendLocalPlanningSnapshot { pub fn planning_request( self, ) -> Result<(PlanningRequest, DeploymentEnvironment), CompileError> { - if self.snapshot_version != 1 && self.snapshot_version != 2 { + if self.snapshot_version != 2 { return Err(CompileError::Snapshot(format!( - "unsupported workload snapshot version {}", + "unsupported workload snapshot version {}; only version 2 is supported", self.snapshot_version ))); } @@ -608,6 +595,7 @@ impl BackendLocalPlanningSnapshot { horizon_seconds: self.implementation.horizon_seconds, costs: self.implementation.lifecycle_costs.clone(), }; + let derived_lifecycle = lifecycle.clone(); let post_asap = crate::planner_selection::keep_pre_asap(&parsed) .map_err(|error| CompileError::Snapshot(format!("query {index}: {error}")))?; canonical_roots.push(Rc::new(parsed)); @@ -636,16 +624,15 @@ impl BackendLocalPlanningSnapshot { .get(&query_string) .cloned() .unwrap_or_else(|| { - vec![WindowImplementationCandidate { - implementation_id: self.implementation.window_implementation_id.clone(), - framework: SummaryWindowFramework::Tumbling, - window_secs: lookback_ms / 1_000, - slide_secs: lookback_ms / 1_000, - layout: asap_types::WindowMaterializationLayout::Pane { - pane_secs: lookback_ms / 1_000, - }, + derived_window_candidates( + &self.implementation.window_implementation_id, + canonical_roots.last().expect("root pushed above"), + lookback_ms, + evaluation_interval_ms, cost, - }] + &derived_lifecycle, + self.implementation.query_staleness_margin_ms, + ) }), runtime_policy: RuntimeRulePolicy::default(), }); @@ -678,6 +665,67 @@ impl BackendLocalPlanningSnapshot { &exact_costs_by_id, self.implementation.erp.as_ref(), )?; + // Derived maintenance currently consumes full, non-overlapping source cohorts. + // Restrict only synthesized candidates; deployment-supplied evidence is authoritative. + for query in &mut queries { + if self + .implementation + .window_candidates + .contains_key(&query.query_string) + { + continue; + } + let states = + collect_selected_materializations(&query.post_asap, true).map_err(|reason| { + CompileError::Query { + query_id: query.query_id.clone(), + reason, + } + })?; + let mut full_windows = BTreeSet::new(); + for state in &states { + if let Some(sources) = immutable_materialization_sources(&state.node) { + full_windows.insert(state.window_secs.unwrap_or(query.window_secs)); + for source in sources { + let (_, window, _) = + selected_input_contract(&source).map_err(|reason| { + CompileError::Query { + query_id: query.query_id.clone(), + reason, + } + })?; + full_windows.insert(window.unwrap_or(query.window_secs)); + } + } + } + let mut seen = BTreeSet::new(); + query.window_implementations.retain_mut(|candidate| { + if !full_windows.contains(&candidate.window_secs) { + return true; + } + if !seen.insert(candidate.window_secs) { + return false; + } + candidate.slide_secs = candidate.window_secs; + candidate.framework = SummaryWindowFramework::Tumbling; + candidate.layout = asap_types::WindowMaterializationLayout::Pane { + pane_secs: candidate.window_secs, + }; + candidate.implementation_id = format!( + "{}-{}s-derived-cohort", + self.implementation.window_implementation_id, candidate.window_secs + ); + candidate.cost = derived_window_cost( + &candidate.cost, + &query.lifecycle, + candidate.window_secs, + candidate.slide_secs, + &candidate.layout, + self.implementation.query_staleness_margin_ms, + ); + true + }); + } // Composable lowering residualizes unsafe leaves individually; retain Planner siblings. Ok(( PlanningRequest { @@ -685,6 +733,16 @@ impl BackendLocalPlanningSnapshot { hybrid_execution: true, materialization_policy: None, query_workload: Some(workload), + synthesized_window_queries: queries + .iter() + .filter(|q| { + !self + .implementation + .window_candidates + .contains_key(&q.query_string) + }) + .map(|q| q.query_id.clone()) + .collect(), queries, evidence: topk_evidence_by_id, exact_composition_costs: exact_costs_by_id, @@ -1058,7 +1116,9 @@ impl PhysicalCompiler { .ok() .flatten() }); - key.is_some_and(|key| policy.contains(&key)) + // Masks enumerate counter/max choices only. Other selected + // summaries remain required by this physical alternative. + key.is_none_or(|key| policy.contains(&key)) }) }) .collect::>(); @@ -1421,6 +1481,18 @@ impl PhysicalCompiler { } } + if environment.target == PhysicalDeploymentTarget::BackendLocalRemoteWrite { + super::pane_reuse::share_additive_panes( + &request, + &mut compiled_materializations, + &mut collector_materializations, + &mut plan_materializations, + &mut node_bindings, + &mut runtime_policies, + &mut lifecycle_estimates, + ); + } + let plan_id = if request.hybrid_execution { use std::hash::{Hash, Hasher}; let mut hash = std::collections::hash_map::DefaultHasher::new(); @@ -1553,7 +1625,31 @@ impl PhysicalCompiler { full_history: false, cumulative_readout: true, }; - let mut entry = if request.hybrid_execution { + // A whole-query native fallback need not be expressible in the local + // residual algebra (for example an ERP-rejected entropy readout). + // Retain its native boundary without discarding other workload roots. + let native_root = request.hybrid_execution + && if let SummaryExpr::KeepPreAsap(expr) = &query.post_asap.expr { + let original = crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .map_err(|error| CompileError::Query { + query_id: query.query_id.clone(), + reason: error.to_string(), + })?; + expr.as_ref() == &original + && crate::query_plan::logical::compile_logical( + query.query_id.clone(), + canonical.clone(), + instant, + FallbackPolicy::ExactBackend, + ) + .is_err() + } else { + false + }; + let mut entry = if request.hybrid_execution && !native_root { crate::query_plan::compile_bound_composable_mapped( query.query_id.clone(), canonical.clone(), @@ -2188,6 +2284,239 @@ fn validate_lifecycle_input( Ok(()) } +/// Price one derived layout from the snapshot's own lifecycle unit costs. +/// +/// `ImplementationCostEvidence` is normally measured evidence, and its +/// `weighted_cost` doc puts pricing update CPU, query-time merges, retained +/// memory, storage, scans and network on the evidence producer. When a +/// snapshot prices no candidate, the control plane becomes that producer for +/// the derived shapes — and it does so without inventing a single magnitude. +/// Every unit cost below is supplied evidence (`LifecycleCostEvidence`, from +/// `implementation.lifecycle_costs`); every multiplier is a structural count +/// that follows from the layout's definition. Nothing here is a measurement. +/// +/// Over `horizon_seconds`, for window `W`, slide `S` and the layout's own +/// shape: +/// +/// - **states created**: a `Pane { P }` seals a state every `P`, a +/// `FullWindow` every `S`. Each one is built once and retired once, so +/// `build` and `retirement` are charged per created state. +/// - **update fanout**: this is the layout's whole point (`worker.rs`'s +/// `stores_full_windows` branch). A pane takes each sample exactly once; a +/// full window takes it into every overlapping window that contains it, +/// `ceil(W / S)` of them. Charged at the supplied ingestion rate. +/// - **finalizations per read**: the mirror image. A full window is read +/// whole; `W / P` panes are composed into one answer. Charged at the +/// query's own evaluation cadence. +/// - **retention**: `retained_state_count`, the same function that fills +/// `num_aggregates_to_retain`, so the quote and the plan cannot disagree. +/// Open worker accumulators are charged separately from published states; +/// updating a state and keeping it resident are different resources. +/// +/// The byte fields are left exactly as the snapshot supplied them: state size +/// needs sketch parameters that do not exist yet at this point, and guessing +/// them would be the fabrication this function otherwise avoids. Only +/// `cpu_cost` and `weighted_cost` are derived, and only those two are read — +/// by `validate_window_implementations` and by the `min_by` that ranks +/// candidates. `model_version` records that the quote is derived. +pub(super) fn derived_window_cost( + template: &ImplementationCostEvidence, + lifecycle: &LifecyclePlanningInput, + window_secs: u64, + slide_secs: u64, + layout: &asap_types::WindowMaterializationLayout, + staleness_margin_ms: u64, +) -> ImplementationCostEvidence { + let costs = &lifecycle.costs; + let horizon = lifecycle.horizon_seconds.max(0.0); + let window = window_secs.max(1) as f64; + let slide = slide_secs.max(1) as f64; + let (seal_interval, update_fanout, finalizations_per_read) = match layout { + asap_types::WindowMaterializationLayout::FullWindow => { + (slide, (window / slide).ceil(), 1.0) + } + asap_types::WindowMaterializationLayout::Pane { pane_secs } => { + let pane = (*pane_secs).max(1) as f64; + (pane, 1.0, (window / pane).ceil()) + } + asap_types::WindowMaterializationLayout::HierarchicalRollup { base_pane_secs, .. } => { + let pane = (*base_pane_secs).max(1) as f64; + (pane, 1.0, (window / pane).ceil()) + } + }; + let states_created = horizon / seal_interval; + let updates = lifecycle.ingestion_rate_per_second.max(0.0) * horizon * update_fanout; + let evaluation_secs = (f64::from(lifecycle.evaluation_interval_ms) / 1_000.0).max(1.0); + let reads = horizon / evaluation_secs; + let retained = retained_state_count( + window_secs.saturating_mul(1_000), + staleness_margin_ms, + slide_secs.saturating_mul(1_000), + layout, + ) as f64; + // Store retention does not include worker accumulators that are still open. + let active = match layout { + asap_types::WindowMaterializationLayout::FullWindow => (window / slide).ceil(), + _ => 1.0, + }; + + let build = costs.build * states_created; + let maintenance = costs.maintenance_per_update * updates; + let read = costs.read * reads * finalizations_per_read; + let retention = costs.retention_per_second * horizon * (retained + active); + let retirement = costs.retirement * states_created; + let cpu_cost = build + maintenance; + let weighted_cost = cpu_cost + read + retention + retirement; + + ImplementationCostEvidence { + model_version: format!( + "{}+derived-window-layout-v1", + template + .model_version + .trim_end_matches("+derived-window-layout-v1") + ), + cpu_cost, + weighted_cost, + ..template.clone() + } +} + +/// Every distinct range-selector window in `expr`, as seconds. +/// +/// A single query can carry several. `sum(sum_over_time(a[1m])) / sum(sum_over_time(b[5m]))` +/// has two, and each one becomes its own materialization with its own window. +/// `time_selection.lookback` is the workload's declared range and is not +/// required to equal any of them. +fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet { + fn visit(expr: &QueryExpr, windows: &mut BTreeSet) { + if let QueryExpr::TimeRange { range, .. } = expr { + let secs = range.as_secs(); + if secs != 0 { + windows.insert(secs); + } + } + match expr { + QueryExpr::PromqlScalarBridge(child) + | QueryExpr::PromqlVectorFromScalar(child) + | QueryExpr::PromqlScalarFromVector(child) + | QueryExpr::PromqlRelabel { child, .. } + | QueryExpr::PromqlSeriesSample { child, .. } + | QueryExpr::Filter { child, .. } + | QueryExpr::Project { child, .. } + | QueryExpr::Aggregate { child, .. } + | QueryExpr::Dedup { child, .. } + | QueryExpr::Sort { child, .. } + | QueryExpr::Limit { child, .. } + | QueryExpr::PromqlSubquery { child, .. } + | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } => visit(child, windows), + QueryExpr::BinaryOp { + lhs: left, + rhs: right, + .. + } + | QueryExpr::Join { left, right, .. } + | QueryExpr::SetOp { left, right, .. } => { + visit(left, windows); + visit(right, windows); + } + _ => {} + } + } + let mut windows = BTreeSet::new(); + visit(expr, &mut windows); + windows +} + +/// Derive and price one implementation per supported layout for each range. +/// Explicit snapshot candidates bypass this path. After logical selection, +/// derived maintenance cohorts are restricted to their supported full windows; +/// raw additive pane producers may subsequently be shared by Planner. +fn derived_window_candidates( + implementation_id: &str, + expr: &QueryExpr, + lookback_ms: u64, + evaluation_interval_ms: u32, + cost: ImplementationCostEvidence, + lifecycle: &LifecyclePlanningInput, + staleness_margin_ms: u64, +) -> Vec { + let mut windows = range_selector_windows_secs(expr); + if windows.is_empty() { + windows.insert(lookback_ms / 1_000); + } + // `window_implementation_id` reaches lifecycle estimates and cost + // manifests, so one label must not describe several shapes. A query with a + // single window keeps the snapshot's identity untouched. + let distinct = windows.len() > 1; + windows + .into_iter() + .flat_map(|window_secs| { + let evaluation_secs = u64::from(evaluation_interval_ms) / 1_000; + let advances_within_window = evaluation_secs != 0 + && evaluation_secs < window_secs + && window_secs.is_multiple_of(evaluation_secs); + let slide_secs = if advances_within_window { + evaluation_secs + } else { + window_secs + }; + let window_label = if distinct { + format!("{implementation_id}-{window_secs}s") + } else { + implementation_id.to_string() + }; + // `Tumbling` pairs only with `Pane` in the validator's + // framework/layout table, so a non-sliding shape has no + // alternative to rank against and keeps its label unchanged. + let layouts: Vec<(String, asap_types::WindowMaterializationLayout)> = + if advances_within_window { + vec![ + ( + format!("{window_label}-pane-{slide_secs}s"), + asap_types::WindowMaterializationLayout::Pane { + pane_secs: slide_secs, + }, + ), + ( + format!("{window_label}-full-window"), + asap_types::WindowMaterializationLayout::FullWindow, + ), + ] + } else { + vec![( + window_label, + asap_types::WindowMaterializationLayout::Pane { + pane_secs: slide_secs, + }, + )] + }; + layouts + .into_iter() + .map(|(id, layout)| WindowImplementationCandidate { + implementation_id: id, + framework: if advances_within_window { + SummaryWindowFramework::Sliding + } else { + SummaryWindowFramework::Tumbling + }, + window_secs, + slide_secs, + cost: derived_window_cost( + &cost, + lifecycle, + window_secs, + slide_secs, + &layout, + staleness_margin_ms, + ), + layout, + }) + .collect::>() + }) + .collect() +} + pub(super) fn validate_window_implementations( query: &PlanningQuery, environment: &DeploymentEnvironment, @@ -2265,7 +2594,7 @@ pub(super) fn validate_window_implementations( Ok(candidates) } -fn retained_state_count( +pub(super) fn retained_state_count( lookback_ms: u64, staleness_margin_ms: u64, slide_ms: u64, @@ -3237,9 +3566,115 @@ fn stable_workload_plan_id( } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; + // Complete deployment quotes must preserve one producer with two window readouts. + #[test] + fn complete_cost_selection_preserves_shared_sum_panes() { + let mut snapshot = planning_snapshot(); + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query("sum_over_time(a[1m]) / sum_over_time(a[10m])".into()); + entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); + let plan = quoted_snapshot(snapshot, false).compile().unwrap(); + assert!(plan.cost_comparison.is_some()); + assert_eq!(plan.precompute_plan.materializations.len(), 1); + let bindings = plan + .query_plan + .entries + .values() + .next() + .unwrap() + .materialization_bindings(); + assert_eq!(bindings[0].materialization, bindings[1].materialization); + assert_eq!( + bindings + .iter() + .filter_map(|b| b.readout_lookback_ms) + .collect::>(), + BTreeSet::from([60_000, 600_000]) + ); + } + + // Synthetic quotes exercise deployment selection in tests, never production defaults. + pub(crate) fn quoted_snapshot( + mut snapshot: BackendLocalPlanningSnapshot, + metricsql: bool, + ) -> BackendLocalPlanningSnapshot { + use super::super::workload_cost::{ + manifest, with_exact_alternative, WorkloadCostEvidence, WorkloadQuote, + }; + let (request, environment) = snapshot.clone().planning_request().unwrap(); + let quotes = with_exact_alternative(request) + .unwrap() + .into_iter() + .enumerate() + .filter_map(|(index, candidate)| { + let plan = if metricsql { + PhysicalCompiler.compile_metricsql(candidate.clone(), environment.clone()) + } else { + PhysicalCompiler.compile(candidate.clone(), environment.clone()) + } + .ok()?; + let manifest = manifest(&plan, &candidate.queries).unwrap(); + Some(WorkloadQuote { + unit_costs: manifest + .components + .keys() + .map(|key| (key.clone(), if index == 0 { 1.0 } else { 1e12 })) + .collect(), + manifest, + executable: true, + }) + }) + .collect(); + snapshot.workload_cost_evidence = Some(WorkloadCostEvidence { + backend_revision: BACKEND_REVISION.into(), + planner_revision: PLANNER_REVISION.into(), + data_snapshot_id: "compiler-unit-fixture".into(), + model_version: "test-only-unit-costs".into(), + observed_at_unix_ms: environment.observed_at_unix_ms, + valid_for_ms: environment.max_evidence_age_ms, + quotes, + }); + snapshot + } + + /// Optional counter masks must retain the workload's mandatory sketch bindings. + #[test] + fn costed_mixed_workload_retains_sketches_and_counter_readouts() { + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let plan = quoted_snapshot(snapshot, false).compile().unwrap(); + assert!(!plan.precompute_plan.materializations.is_empty()); + for entry in plan.query_plan.entries.values() { + assert!(!entry.materialization_bindings().is_empty(), "{entry:#?}"); + } + } + + /// A schema marker cannot opt into a legacy deployment policy. + #[test] + fn only_current_snapshot_schema_is_accepted() { + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + for version in [0, 1, 3] { + let mut old = snapshot.clone(); + old.snapshot_version = version; + assert!(old + .clone() + .planning_request() + .unwrap_err() + .to_string() + .contains("only version 2")); + assert!(old.compile().is_err()); + } + assert!(snapshot.planning_request().is_ok()); + } + #[test] fn installed_partition_must_match_the_bound_dag_reduction() { let mut env = environment(10_000); @@ -3728,6 +4163,7 @@ mod tests { } Ok(PlanningRequest { logical_selection: Vec::new(), + synthesized_window_queries: BTreeSet::new(), hybrid_execution: false, materialization_policy: None, query_workload: None, @@ -3957,7 +4393,7 @@ mod tests { "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile_metricsql().unwrap(); + let plan = quoted_snapshot(snapshot, true).compile_metricsql().unwrap(); assert!(!plan.query_plan.entries.is_empty()); assert!(plan .query_plan @@ -4861,15 +5297,9 @@ mod tests { let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; entry.query = Query("sum(sum_over_time(a[1m])) / sum(sum_over_time(b[5m]))".into()); entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); - let (mut request, env) = snapshot.planning_request().unwrap(); - let query = &mut request.queries[0]; - let mut five_minutes = query.window_implementations[0].clone(); - five_minutes.implementation_id = "five-minute-evidence".into(); - five_minutes.window_secs = 300; - five_minutes.framework = SummaryWindowFramework::Sliding; - five_minutes.slide_secs = 60; - five_minutes.layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 60 }; - query.window_implementations.push(five_minutes); + // The derivation now covers both range selectors, so this no longer + // needs a hand-supplied 5m candidate to keep `b` from falling back. + let (request, env) = snapshot.planning_request().unwrap(); let plan = PhysicalCompiler.compile(request, env).unwrap(); let bindings = plan .query_plan @@ -4890,16 +5320,468 @@ mod tests { ) }) .collect::>(); + // `window_ms` is the stored pane width, `readout_lookback_ms` the + // semantic range. Each operand keeps its own range -- 1m for `a`, 5m + // for `b` -- while both store 10s panes, because the snapshot + // evaluates every 10s and the derivation now covers both selectors. assert_eq!( actual, - BTreeSet::from([("a", 60_000, Some(60_000)), ("b", 60_000, Some(300_000)),]) + BTreeSet::from([("a", 10_000, Some(60_000)), ("b", 10_000, Some(300_000)),]) + ); + assert_eq!(plan.precompute_plan.materializations.len(), 2); + } + + // Derived programs and their raw inputs must keep a runtime-supported cohort. + #[test] + fn derived_window_regression_nested_snapshot() { + for rate in [0.0, 100.0] { + let mut value: Value = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + value["query_workload"]["repeating_queries"][0]["query"] = + json!("quantile(0.9, sum_over_time(m[1m]))"); + value["data_workload"]["ingestion_rate"]["value"] = json!(rate); + value["query_workload"]["data_workload"]["ingestion_rate"]["value"] = json!(rate); + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(value).unwrap(); + let (request, env) = snapshot.planning_request().unwrap(); + let plan = PhysicalCompiler.compile(request, env).unwrap(); + assert!(plan + .precompute_plan + .materializations + .iter() + .any(|m| m.derived_input.is_some())); + assert!(plan + .precompute_plan + .materializations + .iter() + .all(|m| m.window_size == m.slide_interval)); + } + } + + // A full-window producer keeps overlapping accumulators alive even before publication. + #[test] + fn derived_window_regression_resident_cost() { + let template = planning_snapshot().implementation.implementation_cost; + let mut lifecycle = planning_lifecycle(); + lifecycle.costs = LifecycleCostEvidence { + build: 0.0, + maintenance_per_update: 0.0, + read: 0.0, + retention_per_second: 1.0, + retirement: 0.0, + }; + let full = derived_window_cost( + &template, + &lifecycle, + 300, + 30, + &asap_types::WindowMaterializationLayout::FullWindow, + 0, ); + assert!(full.weighted_cost >= lifecycle.horizon_seconds * 11.0); + } + + // Temporal SUM readouts share raw state only for identical source populations. + #[test] + fn derived_window_regression_shared_sum_panes() { + for interval in [10_000, 60_000] { + for (rhs, expected_states) in [("a", 1), ("b", 2), ("a{job=\"x\"}", 2)] { + let mut snapshot = planning_snapshot(); + let query = format!("sum_over_time(a[1m]) / sum_over_time({rhs}[10m])"); + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query(query); + entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); + entry.demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(interval), + evaluation_phase: planner_types::workload::TimestampMs(0), + }; + let (request, env) = snapshot.planning_request().unwrap(); + let plan = PhysicalCompiler.compile(request, env).unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), expected_states); + let entry = plan.query_plan.entries.values().next().unwrap(); + let bindings = entry.materialization_bindings(); + assert_eq!( + bindings + .iter() + .filter_map(|b| b.readout_lookback_ms) + .collect::>(), + BTreeSet::from([60_000, 600_000]) + ); + if rhs == "a" { + assert_eq!(bindings[0].materialization, bindings[1].materialization); + assert_eq!( + plan.precompute_plan.materializations[0].num_aggregates_to_retain, + Some(600_000 / u64::from(interval) + 1) + ); + } + } + } + } + + // Distinct workload entries share one producer and retain both lifecycle consumers. + #[test] + fn shared_panes_preserve_workload_consumers_and_phase() { + for phase in [0, 5_000] { + let mut snapshot = planning_snapshot(); + let entries = snapshot.query_workload.repeating_queries.as_mut().unwrap(); + entries[0].query = Query("sum_over_time(a[1m])".into()); + entries[0].requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); + let mut second = entries[0].clone(); + second.query = Query("sum_over_time(a[10m])".into()); + second.demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(10_000), + evaluation_phase: planner_types::workload::TimestampMs(phase), + }; + entries.push(second); + let (request, env) = snapshot.planning_request().unwrap(); + let plan = PhysicalCompiler.compile(request, env).unwrap(); + assert_eq!( + plan.precompute_plan.materializations.len(), + if phase == 0 { 1 } else { 2 } + ); + if phase == 0 { + assert_eq!(plan.lifecycle_estimates.len(), 1); + let estimate = &plan.lifecycle_estimates[0]; + assert_eq!(estimate.consumer_query_ids.len(), 2); + assert_eq!(estimate.expected_reads, 60.0); + assert_eq!(estimate.expected_updates, 30_000.0); + } + } + } + + // Overflow must fail candidate validation, never turn an expensive layout into a free one. + #[test] + fn derived_cost_overflow_is_rejected() { + let snapshot = planning_snapshot(); + let (mut request, env) = snapshot.planning_request().unwrap(); + let query = &mut request.queries[0]; + let mut lifecycle = query.lifecycle.clone(); + lifecycle.costs.build = f64::MAX; + let candidate = &mut query.window_implementations[0]; + candidate.cost = derived_window_cost( + &candidate.cost, + &lifecycle, + candidate.window_secs, + candidate.slide_secs, + &candidate.layout, + 0, + ); + assert!(validate_window_implementations(query, &env).is_err()); + } + + // Serialized derived quotes become authoritative when a deployment supplies them explicitly. + #[test] + fn explicit_window_quotes_are_not_repriced_for_sharing() { + let mut snapshot = planning_snapshot(); + let query = "sum_over_time(a[1m]) / sum_over_time(a[10m])"; + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query(query.into()); + entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); + let (derived, _) = snapshot.clone().planning_request().unwrap(); + snapshot.implementation.window_candidates.insert( + query.into(), + derived.queries[0].window_implementations.clone(), + ); + let (request, env) = snapshot.planning_request().unwrap(); + assert!(request.synthesized_window_queries.is_empty()); + let plan = PhysicalCompiler.compile(request, env).unwrap(); assert_eq!(plan.precompute_plan.materializations.len(), 2); } - // A filtered denominator is a typed residual while its summary sibling remains installed. + fn planning_lifecycle() -> LifecyclePlanningInput { + planning_snapshot().planning_request().unwrap().0.queries[0] + .lifecycle + .clone() + } + + fn planning_snapshot() -> BackendLocalPlanningSnapshot { + serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap() + } + + // A workload evaluated more often than its window is wide must advance its + // state at that cadence. Planning it as one lookback-wide tumbling window + // answers with results that only change once per window. + #[test] + fn derived_window_candidate_follows_the_evaluation_cadence() { + let cost = planning_snapshot().implementation.implementation_cost; + let expr = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.5, data[5m])", + AccuracyTarget::Exact, + ) + .unwrap(); + let derived = + derived_window_candidates("id", &expr, 300_000, 30_000, cost, &planning_lifecycle(), 0); + // A sliding shape has two legal layouts, so both are offered and the + // cost model picks between them. + assert_eq!( + derived + .iter() + .map(|c| (c.implementation_id.as_str(), c.layout.clone())) + .collect::>(), + vec![ + ( + "id-pane-30s", + asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 } + ), + ( + "id-full-window", + asap_types::WindowMaterializationLayout::FullWindow + ), + ] + ); + let candidate = &derived[0]; + assert_eq!(candidate.framework, SummaryWindowFramework::Sliding); + assert_eq!((candidate.window_secs, candidate.slide_secs), (300, 30)); + assert_eq!( + candidate.layout, + asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 } + ); + } + + // The layout choice is a write-amplification-versus-read-amplification + // trade, and the derived quote has to price it in the right direction: a + // full window takes every sample into all ten overlapping windows but is + // read whole, panes take each sample once but compose ten per read. Which + // wins depends on how hard the source is pushing, so pin the crossover, + // not the magnitudes. #[test] - fn composable_binary_retains_summary_sibling_of_prometheus_filtered_subtree() { + fn derived_window_layout_prices_write_against_read_amplification() { + let cost = planning_snapshot().implementation.implementation_cost; + let expr = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.5, data[5m])", + AccuracyTarget::Exact, + ) + .unwrap(); + let quote = |rate: f64| { + let mut lifecycle = planning_lifecycle(); + lifecycle.ingestion_rate_per_second = rate; + let derived = derived_window_candidates( + "id", + &expr, + 300_000, + 30_000, + cost.clone(), + &lifecycle, + 0, + ); + let weighted = |layout: &asap_types::WindowMaterializationLayout| { + derived + .iter() + .find(|c| c.layout == *layout) + .expect("both layouts offered") + .cost + .weighted_cost + }; + ( + weighted(&asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 }), + weighted(&asap_types::WindowMaterializationLayout::FullWindow), + ) + }; + let (idle_pane, idle_full) = quote(0.0); + assert!( + idle_full < idle_pane, + "with no arriving data the fanout is free and the merges are not: \ + pane {idle_pane} full {idle_full}" + ); + let (busy_pane, busy_full) = quote(100.0); + assert!( + busy_pane < busy_full, + "under load the tenfold update fanout dominates: pane {busy_pane} full {busy_full}" + ); + } + + // A tumbling shape pairs only with `Pane` in the validator's + // framework/layout table, so there is no alternative to price against it. + #[test] + fn tumbling_shapes_have_no_layout_alternative_to_rank() { + let cost = planning_snapshot().implementation.implementation_cost; + let expr = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.5, data[5m])", + AccuracyTarget::Exact, + ) + .unwrap(); + let derived = derived_window_candidates( + "id", + &expr, + 300_000, + 300_000, + cost, + &planning_lifecycle(), + 0, + ); + assert_eq!(derived.len(), 1); + assert_eq!(derived[0].implementation_id, "id"); + assert_eq!(derived[0].framework, SummaryWindowFramework::Tumbling); + } + + // Every shape this function can emit must survive the validator, or a bad + // derivation would reach a plan instead of a compile error. + #[test] + fn derived_window_candidate_shapes_are_accepted_by_validation() { + let snapshot = planning_snapshot(); + let (request, environment) = snapshot.planning_request().unwrap(); + let cost = planning_snapshot().implementation.implementation_cost; + for (lookback_ms, evaluation_ms) in [ + (300_000, 30_000), + (300_000, 300_000), + (300_000, 45_000), + (60_000, 90_000), + ] { + let mut query = request.queries[0].clone(); + query.window_secs = lookback_ms / 1_000; + let expr = crate::query_parser::parse_query_expr_canonical( + &format!("quantile_over_time(0.5, data[{}s])", lookback_ms / 1_000), + AccuracyTarget::Exact, + ) + .unwrap(); + query.window_implementations = derived_window_candidates( + "derived", + &expr, + lookback_ms, + evaluation_ms, + cost.clone(), + &planning_lifecycle(), + 0, + ); + validate_window_implementations(&query, &environment).unwrap_or_else(|error| { + panic!("lookback {lookback_ms} cadence {evaluation_ms}: {error:?}") + }); + } + } + + // A cadence that cannot divide the window has no pane width dividing both, + // and one at or above the window has nothing to slide within. Both keep the + // previous tumbling shape rather than emitting something unschedulable. + #[test] + fn derived_window_candidate_stays_tumbling_without_a_dividing_cadence() { + let cost = planning_snapshot().implementation.implementation_cost; + let expr = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.5, data[5m])", + AccuracyTarget::Exact, + ) + .unwrap(); + for evaluation_ms in [300_000, 450_000, 45_000, 0] { + let derived = derived_window_candidates( + "id", + &expr, + 300_000, + evaluation_ms, + cost.clone(), + &planning_lifecycle(), + 0, + ); + let candidate = &derived[0]; + assert_eq!( + ( + candidate.framework.clone(), + candidate.slide_secs, + candidate.layout.clone() + ), + ( + SummaryWindowFramework::Tumbling, + 300, + asap_types::WindowMaterializationLayout::Pane { pane_secs: 300 } + ), + "cadence {evaluation_ms}" + ); + } + } + + // Priced evidence is the evidence producer's to supply. A snapshot that + // carries its own candidates keeps them verbatim. + #[test] + fn supplied_window_candidates_are_not_replaced_by_the_derivation() { + let mut snapshot = planning_snapshot(); + let query_string = snapshot.query_workload.repeating_queries.as_ref().unwrap()[0] + .query + .0 + .clone(); + let expr = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.99, m[1m])", + AccuracyTarget::Exact, + ) + .unwrap(); + let mut supplied = derived_window_candidates( + "supplied", + &expr, + 60_000, + 60_000, + snapshot.implementation.implementation_cost.clone(), + &planning_lifecycle(), + 0, + ) + .remove(0); + supplied.framework = SummaryWindowFramework::Sliding; + supplied.slide_secs = 20; + supplied.layout = asap_types::WindowMaterializationLayout::Pane { pane_secs: 20 }; + snapshot + .implementation + .window_candidates + .insert(query_string, vec![supplied.clone()]); + let (request, _) = snapshot.planning_request().unwrap(); + assert_eq!(request.queries[0].window_implementations, vec![supplied]); + } + + // End to end: the retained-state count is derived from the pane width, so + // fixing the shape fixes it too. Six 10s panes cover the 1m lookback, plus + // the one still being filled. + #[test] + fn retained_state_count_follows_the_derived_pane_width() { + let snapshot = planning_snapshot(); + let (request, environment) = snapshot.planning_request().unwrap(); + let plan = PhysicalCompiler.compile(request, environment).unwrap(); + assert_eq!( + plan.precompute_plan.materializations[0].num_aggregates_to_retain, + Some(7) + ); + } + + // The reported case: two 5m-lookback quantiles evaluated every 30s. The + // whole chain has to land — sliding framework, 30s panes, and the retained + // count that falls out of the pane width — or the answer only changes once + // every five minutes. + #[test] + fn five_minute_lookback_evaluated_every_thirty_seconds_slides_by_thirty() { + let mut snapshot = planning_snapshot(); + { + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query("quantile_over_time(0.5, data[5m])".into()); + entry.time_selection.lookback = Some(DurationMs(300_000)); + entry.demand = RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(30_000), + evaluation_phase: planner_types::workload::TimestampMs(0), + }; + } + let (request, environment) = snapshot.planning_request().unwrap(); + let plan = PhysicalCompiler.compile(request, environment).unwrap(); + let materialization = &plan.precompute_plan.materializations[0]; + assert_eq!( + ( + materialization.window_size, + materialization.slide_interval, + materialization.window_type, + materialization.window_layout.clone(), + materialization.num_aggregates_to_retain, + ), + ( + 300, + 30, + asap_types::WindowKind::Sliding, + asap_types::WindowMaterializationLayout::Pane { pane_secs: 30 }, + // Ten 30s panes cover the 5m lookback, plus the one still filling. + Some(11), + ) + ); + } + + // A filtered operand gets a summary over its own filtered population, + // with each operand keeping its own range. + #[test] + fn composable_binary_summarizes_each_prometheus_filtered_operand() { use crate::query_plan::{logical::LogicalOperator, QueryPlanNode}; let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!( "../../../docs/examples/asapquery-planning-snapshot.json" @@ -4913,20 +5795,35 @@ mod tests { let plan = PhysicalCompiler.compile(request, env).unwrap(); let query = plan.query_plan.entries.values().next().unwrap(); let bindings = query.materialization_bindings(); - assert_eq!(bindings.len(), 1); - let identity = &plan.summary_catalog.materializations[&bindings[0].materialization]; - let data = &plan.summary_catalog.data_descriptors[&identity.data_descriptor_id]; + // Both operands now hold a summary. The filtered denominator is no + // longer a typed residual: its 5m range has a candidate, so it gets + // its own summary over the filtered population rather than exact + // execution. Nothing about the filter forced the residual -- the + // missing 5m window candidate did, and this test previously pinned + // that artifact as intended behavior. + let bound = bindings + .iter() + .map(|binding| { + let identity = &plan.summary_catalog.materializations[&binding.materialization]; + let data = &plan.summary_catalog.data_descriptors[&identity.data_descriptor_id]; + ( + data.time_series_metric().unwrap(), + data.population_filter_canonical.clone(), + binding.readout_lookback_ms, + ) + }) + .collect::>(); assert_eq!( - (data.time_series_metric().unwrap(), bindings[0].window_ms), - ("a", 60_000) + bound, + BTreeSet::from([ + ("a", String::new(), Some(60_000)), + ("b", "{job!=\"x\"}".to_string(), Some(300_000)), + ]) ); assert!(!query .nodes .values() .any(|node| matches!(node, QueryPlanNode::ExactFallback { .. }))); - assert!(query.nodes.values().any(|node| matches!(node, - QueryPlanNode::Logical { operator: LogicalOperator::ExactSubquery { query }, .. } - if query == "sum_over_time(b{job!=\"x\"}[5m])" || query == "sum(sum_over_time(b{job!=\"x\"}[5m]))"))); assert!(!query.nodes.values().any(|node| matches!( node, QueryPlanNode::Logical { @@ -4934,7 +5831,7 @@ mod tests { .. } ))); - assert_eq!(plan.precompute_plan.materializations.len(), 1); + assert_eq!(plan.precompute_plan.materializations.len(), 2); } #[test] @@ -4993,7 +5890,7 @@ mod tests { "../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - let bundle = snapshot.compile().unwrap(); + let bundle = quoted_snapshot(snapshot, false).compile().unwrap(); let catalog = &bundle.summary_catalog; let mut transmission = bundle.transmission_plan.clone(); transmission.validate_against_catalog(catalog).unwrap(); @@ -5061,7 +5958,7 @@ mod tests { let mut second = entries[0].clone(); second.query = Query("sum(sum_over_time(m[1m])) * 2".into()); entries.push(second); - let bundle = snapshot.compile().unwrap(); + let bundle = quoted_snapshot(snapshot, false).compile().unwrap(); assert_eq!(bundle.query_plan.entries.len(), 2); assert_eq!(bundle.precompute_plan.materializations.len(), 1); let query_plan: QueryPlan = @@ -5403,7 +6300,7 @@ mod tests { .queries .remove(0); let snapshot = BackendLocalPlanningSnapshot { - snapshot_version: 1, + snapshot_version: 2, workload_cost_evidence: None, query_workload, data_workload, @@ -5434,12 +6331,10 @@ mod tests { .as_ref(), Some(&snapshot.query_workload) ); - let first = snapshot - .clone() + let first = quoted_snapshot(snapshot.clone(), false) .compile() .expect("first deterministic plan"); - let second = snapshot - .clone() + let second = quoted_snapshot(snapshot.clone(), false) .compile() .expect("second deterministic plan"); assert_eq!(first.envelope, second.envelope); @@ -5454,7 +6349,9 @@ mod tests { let encoded = serde_json::to_vec(&snapshot).expect("serialize startup snapshot"); let decoded: BackendLocalPlanningSnapshot = serde_json::from_slice(&encoded).expect("deserialize startup snapshot"); - let bundle = decoded.compile().expect("canonical startup planning"); + let bundle = quoted_snapshot(decoded, false) + .compile() + .expect("canonical startup planning"); assert!(bundle.collector_plans.is_empty()); assert!(bundle.transmission_plan.rules.is_empty()); @@ -5542,10 +6439,10 @@ mod tests { let fixture: serde_json::Value = serde_json::from_str(source).expect("fixture JSON"); assert_eq!(encoded, fixture); - snapshot - .clone() - .compile() - .expect("unquoted v1 compatibility startup remains available"); + assert!( + snapshot.clone().compile().is_err(), + "discovery fixtures must be priced before deployment" + ); let (local, env) = snapshot.clone().planning_request().unwrap(); let isolated = PhysicalCompiler.compile(local, env).unwrap(); assert!(!isolated.precompute_plan.materializations.is_empty()); @@ -5578,10 +6475,10 @@ mod tests { include_str!("../../../docs/examples/asapquery-compatibility-demo-snapshot.json"); let snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(source).expect("strict compatibility demo fixture"); - snapshot - .clone() - .compile() - .expect("unquoted v1 compatibility startup remains available"); + assert!( + snapshot.clone().compile().is_err(), + "discovery fixtures must be priced before deployment" + ); let (local, env) = snapshot.clone().planning_request().unwrap(); let isolated = PhysicalCompiler.compile(local, env).unwrap(); assert!(!isolated.precompute_plan.materializations.is_empty()); diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index c41fa2fe..a9e139f1 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -1459,7 +1459,9 @@ mod tests { fixture["query_workload"]["repeating_queries"] = serde_json::json!([query]); let snapshot: crate::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::physical::compiler::tests::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let (mut policy, mut observed) = online_population_fixture(); observed.catalog_generation = plan.summary_catalog.reference().unwrap(); observed.summary_definition_id = diff --git a/control_plane/src/physical/mod.rs b/control_plane/src/physical/mod.rs index 896594da..f26665fb 100644 --- a/control_plane/src/physical/mod.rs +++ b/control_plane/src/physical/mod.rs @@ -27,6 +27,7 @@ pub mod deployment; pub mod deployment_cost; pub mod erp; pub mod executable_binding; +mod pane_reuse; pub mod plan; pub mod plan_cache; pub mod planner; diff --git a/control_plane/src/physical/pane_reuse.rs b/control_plane/src/physical/pane_reuse.rs new file mode 100644 index 00000000..ae34cb15 --- /dev/null +++ b/control_plane/src/physical/pane_reuse.rs @@ -0,0 +1,187 @@ +use super::compiler::{ + derived_window_cost, retained_state_count, CollectorMaterialization, + MaterializationLifecycleEstimate, PlanningRequest, RuntimeRulePolicy, +}; +use planner_types::post_asap::{PostAsapNodeId, SummaryWindowFramework}; +use std::collections::{BTreeMap, BTreeSet}; + +/// Lower Planner's costed pane-reuse groups without changing logical readouts. +/// Only raw additive states are eligible; derived cohorts retain their full-window identity. +#[allow(clippy::too_many_arguments)] +pub(super) fn share_additive_panes( + request: &PlanningRequest, + materializations: &mut [asap_types::PrecomputeMaterialization], + producers: &mut Vec, + plan_producers: &mut [CollectorMaterialization], + bindings: &mut BTreeMap<(usize, PostAsapNodeId), asap_types::PolicyFingerprint>, + policies: &mut BTreeMap, + estimates: &mut BTreeMap, +) { + use asap_aware_mapping::pane_sharing::{select_shared_panes, PaneReuseCandidate}; + use asap_types::{AggregationType, WindowKind, WindowMaterializationLayout}; + let derived_sources = materializations + .iter() + .filter_map(|m| m.derived_input.as_ref()) + .flat_map(|d| d.inputs.iter().map(|id| id.fingerprint())) + .collect::>(); + let mut seen = BTreeSet::new(); + let mut physical = Vec::new(); + let mut offers = Vec::new(); + for m in materializations.iter() { + let old = m.policy_fingerprint(); + if !seen.insert(old) + || m.derived_input.is_some() + || derived_sources.contains(&old) + || !matches!( + m.aggregation_type, + AggregationType::Sum | AggregationType::MultipleSum + ) + { + continue; + } + let WindowMaterializationLayout::Pane { pane_secs } = m.window_layout else { + continue; + }; + if pane_secs == 0 || m.pane_origin_ms.is_none() { + continue; + } + let consumers = bindings + .iter() + .filter(|(_, id)| **id == old) + .map(|((query, _), _)| *query) + .collect::>(); + let Some(&first) = consumers.first() else { + continue; + }; + let query = &request.queries[first]; + // Explicitly priced implementations are not repriced or replaced. + if consumers.iter().any(|index| { + let q = &request.queries[*index]; + q.lifecycle != query.lifecycle + || q.accuracy != query.accuracy + || !request.synthesized_window_queries.contains(&q.query_id) + }) { + continue; + } + let mut canonical = m.clone(); + canonical.window_size = pane_secs; + canonical.slide_interval = pane_secs; + canonical.window_type = WindowKind::Tumbling; + let Some(policy) = policies.get(&old) else { + continue; + }; + let key = ( + canonical.policy_fingerprint(), + serde_json::to_string(&query.lifecycle).unwrap(), + serde_json::to_string(policy).unwrap(), + ); + let mut maintenance = query.lifecycle.clone(); + maintenance.costs.read = 0.0; + let Some(template) = query.window_implementations.first() else { + continue; + }; + let producer_cost = derived_window_cost( + &template.cost, + &maintenance, + m.window_size, + m.slide_interval, + &m.window_layout, + request.query_staleness_margin_ms, + ) + .weighted_cost; + let read_cost = query.lifecycle.costs.read * query.lifecycle.horizon_seconds + / (f64::from(query.lifecycle.evaluation_interval_ms) / 1000.0) + * (m.window_size / pane_secs) as f64 + * consumers.len() as f64; + offers.push(PaneReuseCandidate { + compatibility: key, + lookback_ms: m.window_size.saturating_mul(1000), + producer_cost, + read_cost, + }); + physical.push((old, canonical)); + } + let groups = select_shared_panes(&offers); + let mut target_counts = BTreeMap::new(); + for group in &groups { + *target_counts + .entry(physical[group.members[0]].1.policy_fingerprint()) + .or_insert(0) += 1; + } + let mut replacements = BTreeMap::new(); + for group in groups { + let mut canonical = physical[group.members[0]].1.clone(); + canonical.num_aggregates_to_retain = Some(retained_state_count( + group.lookback_ms, + request.query_staleness_margin_ms, + canonical.slide_interval * 1000, + &canonical.window_layout, + )); + let new = canonical.policy_fingerprint(); + let members = group + .members + .iter() + .map(|index| physical[*index].0) + .collect::>(); + // A physical ID cannot hide a second policy/evidence cohort or an + // independently installed producer that was not offered for sharing. + if target_counts[&new] != 1 + || materializations.iter().any(|m| { + let id = m.policy_fingerprint(); + id == new && !members.contains(&id) + }) + { + continue; + } + let mut combined = estimates[&physical[group.members[0]].0].clone(); + combined.materialization = new.into(); + combined.consumer_query_ids.clear(); + combined.expected_reads = 0.0; + combined.expected_updates = 0.0; + combined.lifecycle_cost = group.cost; + combined.window_implementation_id = format!("shared-pane-{}", new.0); + for index in group.members { + let old = physical[index].0; + if let Some(estimate) = estimates.remove(&old) { + combined + .consumer_query_ids + .extend(estimate.consumer_query_ids); + combined.expected_reads += estimate.expected_reads; + combined.expected_updates = + combined.expected_updates.max(estimate.expected_updates); + } + replacements.insert(old, canonical.clone()); + } + combined.consumer_query_ids.sort(); + combined.consumer_query_ids.dedup(); + estimates.insert(new, combined); + } + for m in materializations { + if let Some(canonical) = replacements.get(&m.policy_fingerprint()) { + *m = canonical.clone(); + } + } + for id in bindings.values_mut() { + if let Some(canonical) = replacements.get(id) { + *id = canonical.policy_fingerprint(); + } + } + for (old, canonical) in &replacements { + if let Some(policy) = policies.remove(old) { + policies.insert(canonical.policy_fingerprint(), policy); + } + } + for producer in producers.iter_mut().chain(plan_producers.iter_mut()) { + if let Some(canonical) = replacements.get(&producer.materialization.fingerprint()) { + let new = canonical.policy_fingerprint(); + producer.materialization = new.into(); + producer.query_id = format!("state-{}", new.0); + producer.window_secs = canonical.window_size; + producer.slide_secs = canonical.slide_interval; + producer.abstract_window_framework = SummaryWindowFramework::Tumbling; + producer.window_implementation_id = estimates[&new].window_implementation_id.clone(); + } + } + let mut seen = BTreeSet::new(); + producers.retain(|producer| seen.insert(producer.materialization)); +} diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index ecf98038..d7a8ea3b 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -991,7 +991,10 @@ mod tests { .count(), 1 ); - assert!(manifest + // Both ranges are summarized now, so nothing in this workload is + // priced as exact backend execution. The 5m operand used to land there + // only because its window had no implementation candidate. + assert!(!manifest .components .values() .any(|demand| demand.implementation.get("location") @@ -1144,9 +1147,8 @@ mod tests { } #[test] - fn v2_snapshot_requires_quotes_and_roundtrips_selection() { + fn snapshot_requires_quotes_and_roundtrips_selection() { let mut snapshot = fixture(); - snapshot.snapshot_version = 2; assert!(snapshot.clone().compile().is_err()); let (_, _, evidence) = quoted(); snapshot.workload_cost_evidence = Some(evidence); diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index a6d63925..bd4037af 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -7230,7 +7230,9 @@ mod catalog_install_tests { "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); PhysicalPlanInstallRequest { summary_catalog: plan.summary_catalog, collector_plans: plan.collector_plans, diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 3c8d7ab4..0fb0c9df 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -2311,7 +2311,9 @@ mod tests { "sum(sum_over_time(m[1m]))".into(); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(snapshot).unwrap(); - let bundle = snapshot.compile().unwrap(); + let bundle = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let mut source_config = bundle.precompute_plan.materializations[0].clone(); // This operator fixture supplies global raw populations; its config // must agree with the explicit Reduce([]) below. @@ -2319,6 +2321,7 @@ mod tests { source_config.grouping_labels = std::iter::empty::().collect(); source_config.window_size = 2; source_config.slide_interval = 2; + source_config.window_type = asap_types::WindowKind::Tumbling; source_config.window_layout = asap_types::aggregation_config::WindowMaterializationLayout::Pane { pane_secs: 1 }; let source_definition = source_config.policy_fingerprint().into(); @@ -3533,7 +3536,12 @@ mod tests { "../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - let mut config = snapshot.compile().unwrap().precompute_plan.materializations[0].clone(); + let mut config = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap() + .precompute_plan + .materializations[0] + .clone(); config.aggregation_type = asap_types::AggregationType::Sum; config.aggregation_sub_type = "sum".into(); config.grouping_labels = ["instance".to_string()].into_iter().collect(); @@ -3589,7 +3597,12 @@ mod tests { "../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - let mut config = snapshot.compile().unwrap().precompute_plan.materializations[0].clone(); + let mut config = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap() + .precompute_plan + .materializations[0] + .clone(); config.aggregation_type = asap_types::AggregationType::DDSketch; config.parameters.clear(); config @@ -3777,7 +3790,9 @@ mod tests { "sum(sum_over_time(m[1m]))".into(); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(snapshot).unwrap(); - let mut bundle = snapshot.compile().unwrap(); + let mut bundle = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let target_config = &bundle.precompute_plan.materializations[0]; let long_step = target_config.window_size.max( target_config.slide_interval * target_config.num_aggregates_to_retain.unwrap_or(1), diff --git a/data_plane/src/precompute_engine/multisource_coordinator.rs b/data_plane/src/precompute_engine/multisource_coordinator.rs index aa60072a..d64cf643 100644 --- a/data_plane/src/precompute_engine/multisource_coordinator.rs +++ b/data_plane/src/precompute_engine/multisource_coordinator.rs @@ -641,7 +641,10 @@ mod tests { wire["query_workload"]["repeating_queries"] = serde_json::json!([query]); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(wire).unwrap(); - let mut plan = snapshot.compile().unwrap().precompute_plan; + let mut plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap() + .precompute_plan; let target = plan .materializations .iter() diff --git a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs index 6bb5877e..2408de04 100644 --- a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs +++ b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs @@ -195,7 +195,9 @@ mod tests { value["query_workload"]["repeating_queries"][0]["query"] = "sum(sum_over_time(m[1m]))".into(); let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(value).unwrap(); - snapshot.compile().unwrap() + crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap() } fn catalog_fixture() -> SummaryCatalog { diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index ccd3a246..516b518d 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -3622,7 +3622,9 @@ mod range_stitch_tests { "../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let mut plan = snapshot.compile().unwrap(); + let mut plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let identity = asap_types::query_plan::canonical_promql("1 + 2").unwrap(); plan.query_plan.entries.insert( asap_types::query_plan::QueryPlan::catalog_key(QueryLanguage::MetricsQl, &identity), diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index 6a4c6268..b91738c3 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs @@ -1160,6 +1160,95 @@ mod tests { assert_eq!(outcome.coverage, Some((2_000, 2_000))); } + // Compile the two readouts, store one pane series, and execute the actual ratio. + #[test] + fn compiled_shared_sum_panes_preserve_each_lookback() { + use control_plane::physical::compiler::{BackendLocalPlanningSnapshot, PhysicalCompiler}; + let mut snapshot: serde_json::Value = serde_json::from_str(include_str!( + "../../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + let entry = &mut snapshot["query_workload"]["repeating_queries"][0]; + entry["query"] = serde_json::json!("sum_over_time(a[1m]) / sum_over_time(a[10m])"); + entry["requirements"]["accuracy"]["explicit"] = serde_json::json!("Exact"); + entry["demand"]["fixed_interval_at"]["interval"] = serde_json::json!(60_000); + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(snapshot).unwrap(); + let (request, env) = snapshot.planning_request().unwrap(); + let plan = PhysicalCompiler.compile(request, env).unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 1); + let config = &plan.precompute_plan.materializations[0]; + let policy = config.policy_fingerprint(); + let idx = SketchStore::new(); + idx.register(SketchInstanceMetadata { + sid: 7, + metric_name: "a".into(), + group_by_keys: Default::default(), + capability: Some(Capability::ExactAgg(asap_types::AggregationType::Sum)), + agg_kind: AggKind::ExactAgg { + agg_type: asap_types::AggregationType::Sum, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: policy, + }); + for pane in 0..11 { + idx.append_precompute( + 7, + BTreeMap::new(), + (pane * 60_000, (pane + 1) * 60_000), + Box::new( + crate::precompute_engine::operators::SumAccumulator::with_sum( + (pane + 1) as f64, + ), + ), + ); + } + let entry = plan.query_plan.entries.values().next().unwrap(); + for (now, expected) in [(600_000, 10.0 / 55.0), (660_000, 11.0 / 65.0)] { + let (outcome, stats) = super::super::logical_dag::execute_installed( + entry, + &BTreeMap::new(), + now, + |root, at| { + let mut subtree = entry.clone(); + subtree.root = root; + let reachable = subtree.topological_order().unwrap(); + subtree.nodes.retain(|id, _| reachable.contains(id)); + subtree.instant.lookback_ms = subtree.materialization_bindings()[0] + .readout_lookback_ms + .unwrap(); + super::super::live_serve::serve_instant_from_query_plan(&idx, &subtree, at) + .map(|(result, _)| { + use crate::query_engines::query_result::{InstantVectorElement, QueryResult}; + let values = result.series.into_iter().map(|(labels, samples)| { + let (keys, values) = labels.into_iter().unzip(); + InstantVectorElement::new(crate::storage_engines::types::KeyByLabelValues::new_with_labels(values), samples.last().unwrap().1) + .with_label_keys_override(keys) + }).collect(); + QueryResult::vector(values, at) + }) + .map_err(|error| { + crate::query_engines::EngineError::capability_miss( + "test", + format!("{error:?}"), + ) + }) + }, + ) + .unwrap(); + let crate::query_engines::query_result::QueryResult::Vector(outcome) = outcome else { + panic!("expected vector"); + }; + assert_eq!(stats.summary_readout_evaluations, 2); + assert_eq!(outcome.values.len(), 1); + assert!((outcome.values[0].value - expected).abs() < 1e-12); + } + } + #[test] fn repeated_multi_pane_reads_exclude_expired_state_and_reject_gaps() { let idx = SketchStore::new(); diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs index 7e7ac4d6..9f01d1d0 100644 --- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs +++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs @@ -742,7 +742,9 @@ mod tests { "../../../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let mut first = plan.precompute_plan.materializations[0].clone(); first.aggregation_type = asap_types::AggregationType::Sum; first.aggregation_sub_type = "sum".into(); @@ -851,7 +853,9 @@ mod tests { "../../../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let mut first = plan.precompute_plan.materializations[0].clone(); first.aggregation_type = asap_types::AggregationType::Sum; first.aggregation_sub_type = "sum".into(); @@ -1091,7 +1095,9 @@ mod tests { fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let source = plan .precompute_plan .materializations @@ -1438,7 +1444,9 @@ mod tests { "../../../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let mut source_config = plan.precompute_plan.materializations[0].clone(); source_config.aggregation_type = asap_types::AggregationType::Sum; source_config.aggregation_sub_type = "sum".into(); diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index d77030e2..63c6070e 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -3789,7 +3789,9 @@ mod tests { "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); let store = SketchStore::new(); store @@ -3862,7 +3864,9 @@ mod tests { "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); let store = SketchStore::new(); store @@ -5020,7 +5024,9 @@ mod tests { "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); let metadata = meta_with_policy(507, fingerprint); let record = SidMetaRecord::new( @@ -5061,7 +5067,9 @@ mod tests { "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); let definition = fingerprint.into(); let mut next_catalog = plan.summary_catalog.clone(); @@ -5160,7 +5168,9 @@ mod tests { "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); let directory = tempfile::tempdir().unwrap(); let store = SketchStore::new(); @@ -5246,7 +5256,9 @@ mod tests { "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); let directory = tempfile::tempdir().unwrap(); { @@ -5354,7 +5366,9 @@ mod tests { "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); let directory = tempfile::tempdir().unwrap(); let disk = directory.path().to_path_buf(); @@ -5436,7 +5450,9 @@ mod tests { "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) .unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = crate::tests::test_utilities::planning::quoted_snapshot(snapshot, false) + .compile() + .unwrap(); let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); let definition_id = SummaryDefinitionId::from(fingerprint); let producers = BTreeMap::from([(definition_id, "producer-a".to_string())]); diff --git a/data_plane/src/tests/test_utilities/mod.rs b/data_plane/src/tests/test_utilities/mod.rs index 2b419d48..2f48c9da 100644 --- a/data_plane/src/tests/test_utilities/mod.rs +++ b/data_plane/src/tests/test_utilities/mod.rs @@ -10,3 +10,5 @@ pub mod engine_factories; pub mod timing; pub use engine_factories::*; + +pub mod planning; diff --git a/data_plane/src/tests/test_utilities/planning.rs b/data_plane/src/tests/test_utilities/planning.rs new file mode 100644 index 00000000..8e6bb709 --- /dev/null +++ b/data_plane/src/tests/test_utilities/planning.rs @@ -0,0 +1,47 @@ +//! Synthetic complete quotes for deployment fixtures; never used in production. +use control_plane::physical::compiler::{ + BackendLocalPlanningSnapshot, PhysicalCompiler, BACKEND_REVISION, PLANNER_REVISION, +}; + +pub(crate) fn quoted_snapshot( + mut snapshot: BackendLocalPlanningSnapshot, + metricsql: bool, +) -> BackendLocalPlanningSnapshot { + use control_plane::physical::workload_cost::{ + manifest, with_exact_alternative, WorkloadCostEvidence, WorkloadQuote, + }; + let (request, environment) = snapshot.clone().planning_request().unwrap(); + let quotes = with_exact_alternative(request) + .unwrap() + .into_iter() + .enumerate() + .filter_map(|(index, candidate)| { + let plan = if metricsql { + PhysicalCompiler.compile_metricsql(candidate.clone(), environment.clone()) + } else { + PhysicalCompiler.compile(candidate.clone(), environment.clone()) + } + .ok()?; + let manifest = manifest(&plan, &candidate.queries).unwrap(); + Some(WorkloadQuote { + unit_costs: manifest + .components + .keys() + .map(|key| (key.clone(), if index == 0 { 1.0 } else { 1e12 })) + .collect(), + manifest, + executable: true, + }) + }) + .collect(); + snapshot.workload_cost_evidence = Some(WorkloadCostEvidence { + backend_revision: BACKEND_REVISION.into(), + planner_revision: PLANNER_REVISION.into(), + data_snapshot_id: "data-plane-unit-fixture".into(), + model_version: "test-only-unit-costs".into(), + observed_at_unix_ms: environment.observed_at_unix_ms, + valid_for_ms: environment.max_evidence_age_ms, + quotes, + }); + snapshot +} diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 54cad0d8..e6953605 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -28,6 +28,60 @@ mod durable_summary_process; #[path = "support/immutable_maintenance_process.rs"] mod immutable_maintenance_process; +// Test-only quotes preserve the fixture's local candidate without a production bypass. +fn quote_snapshot_for_test( + snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot, +) -> control_plane::physical::compiler::BackendLocalPlanningSnapshot { + quote_snapshot_for_frontend_test(snapshot, false) +} + +fn quote_snapshot_for_frontend_test( + mut snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot, + metricsql: bool, +) -> control_plane::physical::compiler::BackendLocalPlanningSnapshot { + use control_plane::physical::{ + compiler::{PhysicalCompiler, BACKEND_REVISION, PLANNER_REVISION}, + workload_cost::{self, WorkloadCostEvidence, WorkloadQuote}, + }; + let (request, environment) = snapshot.clone().planning_request().unwrap(); + let mut preferred = true; + let quotes = workload_cost::with_exact_alternative(request) + .unwrap() + .into_iter() + .enumerate() + .filter_map(|(_index, candidate)| { + let plan = if metricsql { + PhysicalCompiler.compile_metricsql(candidate.clone(), environment.clone()) + } else { + PhysicalCompiler.compile(candidate.clone(), environment.clone()) + } + .ok()?; + let unit_cost = if preferred { 1.0 } else { 1e12 }; + preferred = false; + let manifest = workload_cost::manifest(&plan, &candidate.queries).unwrap(); + Some(WorkloadQuote { + unit_costs: manifest + .components + .keys() + .map(|key| (key.clone(), unit_cost)) + .collect(), + manifest, + executable: true, + }) + }) + .collect(); + snapshot.workload_cost_evidence = Some(WorkloadCostEvidence { + backend_revision: BACKEND_REVISION.into(), + planner_revision: PLANNER_REVISION.into(), + data_snapshot_id: "process-fixture".into(), + model_version: "test-only-unit-costs".into(), + observed_at_unix_ms: environment.observed_at_unix_ms, + valid_for_ms: environment.max_evidence_age_ms, + quotes, + }); + snapshot +} + struct ChildGuard(Child); impl Drop for ChildGuard { @@ -1143,15 +1197,18 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() let backend_port = unused_port(); let output_dir = tempfile::tempdir().expect("backend output directory"); - let snapshot = concat!( - env!("CARGO_MANIFEST_DIR"), - "/../docs/examples/asapquery-compatibility-demo-snapshot.json" - ); + let fixture = serde_json::from_str(include_str!( + "../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let priced = quote_snapshot_for_test(fixture); + let snapshot = output_dir.path().join("snapshot.json"); + std::fs::write(&snapshot, serde_json::to_vec(&priced).unwrap()).unwrap(); let child = Command::new(env!("CARGO_BIN_EXE_data_plane")) .arg("--profile") .arg("asapquery") .arg("--planning-snapshot") - .arg(snapshot) + .arg(&snapshot) .arg("--prometheus-server") .arg(format!("http://{fallback_address}")) .arg("--forward-unsupported-queries") @@ -1588,7 +1645,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() .as_array() .expect("materialization statuses"); let planned_snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = - serde_json::from_str(&std::fs::read_to_string(snapshot).unwrap()).unwrap(); + serde_json::from_str(&std::fs::read_to_string(&snapshot).unwrap()).unwrap(); let planned = planned_snapshot.compile().unwrap(); // Every selected state must be serving; the Planner may share or separate // physical populations, so compare identities rather than a frozen count. diff --git a/data_plane/tests/support/distinct_planning_process.rs b/data_plane/tests/support/distinct_planning_process.rs index 123624db..a67a197a 100644 --- a/data_plane/tests/support/distinct_planning_process.rs +++ b/data_plane/tests/support/distinct_planning_process.rs @@ -23,10 +23,11 @@ async fn distinct_range_uses_planner_selected_hll_and_source_labels() { entry["query"] = QUERY.into(); entry["requirements"]["accuracy"] = serde_json::json!({"explicit": {"Epsilon": 0.05}}); fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); - let plan = serde_json::from_value::(fixture.clone()) - .unwrap() - .compile() - .unwrap(); + let plan = quote_snapshot_for_test( + serde_json::from_value::(fixture.clone()).unwrap(), + ) + .compile() + .unwrap(); assert_eq!(plan.precompute_plan.materializations.len(), 1); assert_eq!( plan.precompute_plan.materializations[0].aggregation_type, @@ -38,7 +39,8 @@ async fn distinct_range_uses_planner_selected_hll_and_source_labels() { ); let output = tempfile::tempdir().unwrap(); let path = output.path().join("planning.json"); - std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap(); + let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap()); + std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap(); let port = unused_port(); let mut vm_port = unused_port(); while vm_port == port { @@ -80,11 +82,10 @@ async fn distinct_range_uses_planner_selected_hll_and_source_labels() { // Source syntax uses the shared parser fork; serving semantics and exact // routing belong to the MetricsQL adapter and its installed query entries. let snapshot = serde_json::from_value::(fixture).unwrap(); - let (mut request, mut environment) = snapshot.planning_request().unwrap(); - request.hybrid_execution = false; - environment.plan_version = 2; - let compiled = control_plane::physical::compiler::PhysicalCompiler - .compile_metricsql(request, environment) + let mut snapshot = snapshot; + snapshot.environment.plan_version = 2; + let compiled = quote_snapshot_for_frontend_test(snapshot, true) + .compile_metricsql() .unwrap(); let identity = serde_json::json!({"plan_id": compiled.envelope.plan_id, "plan_version": compiled.envelope.plan_version}); let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { diff --git a/data_plane/tests/support/durable_summary_process.rs b/data_plane/tests/support/durable_summary_process.rs index b37d39c8..626392ae 100644 --- a/data_plane/tests/support/durable_summary_process.rs +++ b/data_plane/tests/support/durable_summary_process.rs @@ -15,7 +15,7 @@ async fn persisted_summary_restarts_without_live_reregistration() { serde_json::json!([fixture["query_workload"]["repeating_queries"][2].clone()]); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = quote_snapshot_for_test(snapshot).compile().unwrap(); let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { summary_catalog: plan.summary_catalog, collector_plans: plan.collector_plans, diff --git a/data_plane/tests/support/erp_planning_process.rs b/data_plane/tests/support/erp_planning_process.rs index aa0cf508..6e88beac 100644 --- a/data_plane/tests/support/erp_planning_process.rs +++ b/data_plane/tests/support/erp_planning_process.rs @@ -104,7 +104,7 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write() )); let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture.clone()).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = quote_snapshot_for_test(snapshot).compile().unwrap(); assert_eq!( plan.precompute_plan.materializations.len(), 1, @@ -131,7 +131,8 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write() ); let output = tempfile::tempdir().unwrap(); let path = output.path().join("planning.json"); - std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap(); + let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap()); + std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap(); let port = unused_port(); let mut child = ChildGuard( Command::new(env!("CARGO_BIN_EXE_data_plane")) diff --git a/data_plane/tests/support/immutable_maintenance_process.rs b/data_plane/tests/support/immutable_maintenance_process.rs index eb191c0a..f105cdfa 100644 --- a/data_plane/tests/support/immutable_maintenance_process.rs +++ b/data_plane/tests/support/immutable_maintenance_process.rs @@ -35,7 +35,7 @@ async fn run_maintenance_process(multi_source: bool, distinct_groups: bool) { fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(fixture.clone()).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = quote_snapshot_for_test(snapshot).compile().unwrap(); assert_eq!( plan.precompute_plan.materializations.len(), if multi_source { 3 } else { 2 } @@ -303,12 +303,15 @@ async fn run_maintenance_process(multi_source: bool, distinct_groups: bool) { // singleton-derived output while new source populations can arrive. let mut next_fixture = fixture.clone(); next_fixture["environment"]["plan_version"] = 2.into(); - let next = serde_json::from_value::< - control_plane::physical::compiler::BackendLocalPlanningSnapshot, - >(next_fixture) - .unwrap() - .compile() - .unwrap(); + let next = + quote_snapshot_for_test( + serde_json::from_value::< + control_plane::physical::compiler::BackendLocalPlanningSnapshot, + >(next_fixture) + .unwrap(), + ) + .compile() + .unwrap(); let next_install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { summary_catalog: next.summary_catalog, collector_plans: next.collector_plans, diff --git a/data_plane/tests/support/univmon_erp_process.rs b/data_plane/tests/support/univmon_erp_process.rs index 9c933b8c..7c496d4a 100644 --- a/data_plane/tests/support/univmon_erp_process.rs +++ b/data_plane/tests/support/univmon_erp_process.rs @@ -125,7 +125,7 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { "runtime": {"allowed_algorithms": ["Hll", "Kll", "UnivMon"], "max_memory_bytes": null} }); let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture.clone()).unwrap(); - let plan = snapshot.compile().unwrap(); + let plan = quote_snapshot_for_test(snapshot).compile().unwrap(); eprintln!( "UNIVMON_PLANNED {}", serde_json::json!({"query_plan": plan.query_plan, "materializations": plan.precompute_plan.materializations, "lifecycle_estimates": plan.lifecycle_estimates, "executable_dags": plan.precompute_plan.executable_dags, "observation": observation}) @@ -148,10 +148,11 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { .unwrap() .remove("max_frequency_entropy_absolute_bits_error"); } - let missing = serde_json::from_value::(missing_entropy) - .unwrap() - .compile() - .unwrap(); + let missing = quote_snapshot_for_test( + serde_json::from_value::(missing_entropy).unwrap(), + ) + .compile() + .unwrap(); use control_plane::query_plan::{QueryPlanNode, QueryReadout}; assert!(missing .query_plan @@ -210,7 +211,8 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { }); let output = tempfile::tempdir().unwrap(); let path = output.path().join("planning.json"); - std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap(); + let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap()); + std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap(); let port = unused_port(); let mut child = ChildGuard( Command::new(env!("CARGO_BIN_EXE_data_plane")) @@ -326,7 +328,7 @@ async fn measured_readout_evidence_selects_and_executes_univmon() { .unwrap() .invalid_reason .is_none()); - let replanned = live_snapshot.compile().unwrap(); + let replanned = quote_snapshot_for_test(live_snapshot).compile().unwrap(); assert!( replanned .precompute_plan diff --git a/demos/asapquery/run.sh b/demos/asapquery/run.sh index addad6b5..95546995 100755 --- a/demos/asapquery/run.sh +++ b/demos/asapquery/run.sh @@ -11,6 +11,7 @@ PROM_CONTAINER="${RUN_ID}-prometheus" PUSH_CONTAINER="${RUN_ID}-pushgateway" EVIDENCE_DIR="${ASAPQUERY_DEMO_EVIDENCE_DIR:-${REPO_DIR}/target/asapquery-demo-evidence}" BACKEND_PID="" +: "${ASAPQUERY_PLANNING_SNAPSHOT:?Set ASAPQUERY_PLANNING_SNAPSHOT to a snapshot with complete workload cost evidence}" cleanup() { if [[ -n "${BACKEND_PID}" ]]; then @@ -52,7 +53,7 @@ curl -fsS http://127.0.0.1:19090/-/healthy >/dev/null "${REPO_DIR}/target/debug/data_plane" \ --profile asapquery \ - --planning-snapshot "${REPO_DIR}/docs/examples/asapquery-compatibility-demo-snapshot.json" \ + --planning-snapshot "${ASAPQUERY_PLANNING_SNAPSHOT}" \ --prometheus-server http://127.0.0.1:19090 \ --forward-unsupported-queries \ --http-port 19091 \ diff --git a/docs/developer_docs/planning/shared-window-panes.md b/docs/developer_docs/planning/shared-window-panes.md new file mode 100644 index 00000000..15eea28b --- /dev/null +++ b/docs/developer_docs/planning/shared-window-panes.md @@ -0,0 +1,47 @@ +# Shared additive window panes + +The fallback window provider prices each query range. After logical selection, +raw inputs of derived maintenance programs use full, non-overlapping windows, +matching the current maintenance executor. Explicit `window_candidates` remain +authoritative and are not rewritten. + +For backend-local raw SUM states, the backend offers compatible selected pane +producers to ASAPPlanner's `pane_sharing::select_shared_panes`. The compatibility +key preserves source, population predicate, projection, accumulator parameters, +grouping, partitioning, pane width/origin, runtime policy and lifecycle evidence. +Planner compares independent producers with one producer retained for the longest +lookback, charging every readout. Only beneficial groups are installed. + +The shared state's physical window is one pane. Logical readout windows stay on +their original QueryPlan bindings. Retention uses the largest bound lookback. +For `sum_over_time(a[1m]) / sum_over_time(a[10m])` evaluated every minute, one +60-second producer retains 11 published states; readouts merge one and ten panes. +Different metrics, predicates or phases remain separate. Derived inputs, explicit +window quotes, non-additive state and independently selected FullWindow layouts +are not rewritten. This pass reuses compatible selected panes; it does not search +all possible pane widths or re-optimize FullWindow choices jointly. + +Layout pricing charges both published retained states and open worker +accumulators. A full window of width W and step S keeps ceil(W/S) open states; +a pane producer keeps one. These residency costs are separate from update CPU. +Nonfinite arithmetic remains invalid evidence rather than becoming a zero quote. + +## Cross-repository validation + +The backend pins Planner commit `ca7546de792d74aee8231e9a1100ca893d9e86d3`, which provides +`pane_sharing::select_shared_panes`. Normal builds use the Git dependency. +For coordinated local development, the optional validation script exports only +the optimizer crate while retaining the pinned IR/frontend revision and restores +Cargo.lock after the run. + +```sh +python3 tools/test_shared_panes.py --planner /path/to/ASAPPlanner -- \ + test -p control_plane +python3 tools/test_shared_panes.py --planner /path/to/ASAPPlanner \ + --sketchlib /path/to/compatible/asap_sketchlib -- \ + test -p data_plane --lib compiled_shared_sum_panes_preserve_each_lookback +``` + +Use `--toolchain 1.98.0` if needed in the development environment. The data-plane +checkout requires sketchlib's standard-update guard and interpolated quantile +interfaces; validation used revision `8c03d7c` for those existing dependencies. diff --git a/docs/evaluation/e2e-physical-dag.md b/docs/evaluation/e2e-physical-dag.md index c17d609f..1a3c2807 100644 --- a/docs/evaluation/e2e-physical-dag.md +++ b/docs/evaluation/e2e-physical-dag.md @@ -39,10 +39,9 @@ Planner owns query semantics, summary families, parameters and candidate selection. ERP can affect supported evidence-based choices, but supplying an artifact does not prove that it was eligible or used. Freshness, source/update semantics, parameters and accuracy constraints still apply. The checked-in demo -is not an empirical-ERP benchmark. For measured cost selection use the version-2 -snapshot workflow in [execution calibration](../../tools/o11y-execution/CALIBRATION.md). -`compile_workload_artifact` deliberately requires that version; do not bypass it -by relabeling demo costs as measured evidence. +is not an empirical-ERP benchmark. For deployment, use the priced snapshot workflow in [execution calibration](../../tools/o11y-execution/CALIBRATION.md). +`compile_workload_artifact` requires complete workload quotes; do not relabel +demo costs as measured evidence. For the existing observation → ERP-selected KLL → installed HTTP correctness fixture, see [ERP process validation](../developer_docs/erp-process-validation.md): @@ -62,8 +61,8 @@ Run the normal snapshot compiler, without selecting a candidate index or substituting a preferred sketch family: ```bash -cargo run --locked -p control_plane --example inspect_physical_dag -- \ - docs/examples/asapquery-compatibility-demo-snapshot.json \ +cargo run --locked -p control_plane --example compile_workload_artifact -- \ + "$ASAPQUERY_PLANNING_SNAPSHOT" \ > target/physical-dag-inspection/selected.json jq '.install_request' target/physical-dag-inspection/selected.json \ > target/physical-dag-inspection/physical-plan.json @@ -71,10 +70,12 @@ jq '.install_request | {summary_catalog, precompute_plan, query_plan}' \ target/physical-dag-inspection/selected.json ``` -`inspect_physical_dag` calls the same `BackendLocalPlanningSnapshot::compile` -entry point used by startup. It accepts the checked-in demonstration snapshot -and labels its output `inspection_only`. `erp_input_supplied` reports only input -presence. A null `cost_comparison` must not be interpreted as a measured win. +`compile_workload_artifact` calls the same evidence-required snapshot compiler +used by startup. Set `ASAPQUERY_PLANNING_SNAPSHOT` to a priced snapshot prepared +using the [cost evidence workflow](../examples/workload-cost-evidence.md). +The checked-in unquoted templates support candidate discovery only. The output +includes the selected plan, logical selection trace and complete cost comparison. +MetricsQL compilation requires quotes collected for that frontend. The compiler derives sibling plans from the selected post-ASAP DAG: @@ -95,6 +96,7 @@ participate in identity, cost and coverage checks. For the simplest live run: ```bash +export ASAPQUERY_PLANNING_SNAPSHOT=/absolute/path/priced-snapshot.json ./scripts/e2e.sh asapquery-demo ``` @@ -116,7 +118,7 @@ Then keep the backend running for inspection: ```bash cargo run --locked -p data_plane -- \ --profile asapquery \ - --planning-snapshot docs/examples/asapquery-compatibility-demo-snapshot.json \ + --planning-snapshot "$ASAPQUERY_PLANNING_SNAPSHOT" \ --prometheus-server http://127.0.0.1:9090 \ --forward-unsupported-queries --http-port 9091 \ --output-dir target/physical-dag-inspection/runtime @@ -242,7 +244,7 @@ when comparing, `comparison.json`. A finished run is not proof of benefit. | Backend | Drop-in surface and current boundary | | --- | --- | | Prometheus | Remote Write v1 samples plus PromQL instant/range HTTP; keep Prometheus as the exact fallback and raw-data authority. The strict `asapquery` profile has explicit startup exclusions. | -| VictoriaMetrics | The broader backend has a MetricsQL compile/adapter path. `inspect_physical_dag SNAPSHOT.json --metricsql` uses it. This does not turn the strict Prometheus profile into a general VM replacement; counter boundary semantics and unsupported expressions must retain exact routing. | +| VictoriaMetrics | The broader backend has a MetricsQL compile/adapter path. `compile_workload_artifact SNAPSHOT.json --metricsql` uses it. This does not turn the strict Prometheus profile into a general VM replacement; counter boundary semantics and unsupported expressions must retain exact routing. | | ClickHouse | The broader backend has a SQL workload compiler and typed exact/relational execution. `compile_clickhouse_workload` reads its own `ClickHouseSqlAutomaticWorkload` JSON from stdin; it does not consume the Prometheus snapshot. List/Map/Tuple support does not imply arbitrary lambdas/counter SQL or full protocol compatibility. External-only DAG coverage is not summary acceleration. | Use the existing process suites to validate a specific supported protocol shape; diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json index e345114c..8641f93b 100644 --- a/docs/examples/asapquery-compatibility-demo-snapshot.json +++ b/docs/examples/asapquery-compatibility-demo-snapshot.json @@ -1,5 +1,5 @@ { - "snapshot_version": 1, + "snapshot_version": 2, "query_workload": { "language": "promql", "query_batch": null, diff --git a/docs/examples/asapquery-planning-snapshot.json b/docs/examples/asapquery-planning-snapshot.json index 99c2a284..cd63e8bc 100644 --- a/docs/examples/asapquery-planning-snapshot.json +++ b/docs/examples/asapquery-planning-snapshot.json @@ -1,5 +1,5 @@ { - "snapshot_version": 1, + "snapshot_version": 2, "query_workload": { "language": "promql", "query_batch": null, diff --git a/docs/examples/workload-cost-evidence.md b/docs/examples/workload-cost-evidence.md index 612f58e7..64e60498 100644 --- a/docs/examples/workload-cost-evidence.md +++ b/docs/examples/workload-cost-evidence.md @@ -1,8 +1,10 @@ # Complete workload cost evidence -The migrated, evidence-required startup profile uses `snapshot_version: 2`. -Version 1 and live requests without `workload_cost_evidence` remain compatibility -paths: their lifecycle estimates are not complete workload costs. +Planning snapshots use one schema, `snapshot_version: 2`. Version 1 is rejected. +Candidate discovery may omit `workload_cost_evidence`; compiling a deployable +snapshot requires complete, valid quotes and selects by complete workload cost. +There is no unquoted snapshot deployment path. The checked-in JSON examples are +discovery templates, not ready-to-deploy plans. ## Workflow diff --git a/tools/test_shared_panes.py b/tools/test_shared_panes.py new file mode 100644 index 00000000..30b3cc73 --- /dev/null +++ b/tools/test_shared_panes.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Validate the unmerged Planner/backend pair without changing immutable IR pins.""" +import argparse +import json +from pathlib import Path +import re +import shutil +import subprocess +import tempfile + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--planner", required=True, type=Path) + parser.add_argument("--sketchlib", type=Path) + parser.add_argument("--toolchain") + parser.add_argument("cargo_args", nargs=argparse.REMAINDER) + args = parser.parse_args() + backend = Path(__file__).resolve().parents[1] + manifest = (backend / "control_plane/Cargo.toml").read_text() + declaration = re.search(r"^planner-types(?:\.workspace)?\s*=\s*(.+)$", manifest, re.MULTILINE) + if declaration is None: + raise RuntimeError("planner-types dependency is missing") + if re.search(r"workspace\s*=\s*true", declaration.group(0)): + manifest = (backend / "Cargo.toml").read_text() + declaration = re.search(r"^planner-types\s*=\s*(.+)$", manifest, re.MULTILINE) + revision_match = re.search(r'rev\s*=\s*"([0-9a-f]+)"', declaration.group(1)) if declaration else None + if revision_match is None: + raise RuntimeError("planner-types must declare a pinned Git revision") + revision = revision_match.group(1) + lock = backend / "Cargo.lock" + original_lock = lock.read_bytes() + with tempfile.TemporaryDirectory(prefix="asap-pane-reuse-") as temporary: + root = Path(temporary) + mapping = root / "asap-aware-mapping" + shutil.copytree(args.planner.resolve() / "crates/asap-aware-mapping", mapping) + cargo_toml = mapping / "Cargo.toml" + source = cargo_toml.read_text() + old = 'asap-types = { path = "../types" }' + if old not in source: + raise RuntimeError("unexpected Planner dependency declaration") + cargo_toml.write_text(source.replace(old, 'asap-types = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "' + revision + '" }')) + config = root / "validation.toml" + text = "" + if args.sketchlib: + text += "paths = [" + json.dumps(str(args.sketchlib.resolve())) + "]\n" + text += '[patch."https://github.com/ProjectASAP/ASAPPlanner"]\nasap-aware-mapping = { path = ' + json.dumps(str(mapping)) + ' }\n' + config.write_text(text) + command = ["cargo"] + if args.toolchain: + command.append("+" + args.toolchain) + cargo_args = args.cargo_args + if cargo_args[:1] == ["--"]: + cargo_args = cargo_args[1:] + command += ["--config", str(config)] + (cargo_args or ["test", "-p", "control_plane"]) + try: + return subprocess.call(command, cwd=backend) + finally: + lock.write_bytes(original_lock) + + +if __name__ == "__main__": + raise SystemExit(main())