diff --git a/README.md b/README.md index f9391cbc1..ae06b77f7 100644 --- a/README.md +++ b/README.md @@ -302,13 +302,16 @@ dot -Tsvg target/readme-evidence/selected.dot \ -o target/readme-evidence/selected.svg ``` -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. +Candidate discovery and deployment accept snapshots without external workload +quotes. The backend combines applicable ERP resources or analytical estimates +with data size, query frequency and physical-plan structure, then compares +complete candidate costs, including exact fallback. The selected plan includes +the resource breakdown and assumptions in `cost_comparison`. + +Point `ASAPQUERY_PLANNING_SNAPSHOT` to your workload snapshot with current data +and capability inputs. Optional calibrated provider quotes can override automatic +costing through the [cost evidence workflow](docs/examples/workload-cost-evidence.md). +Materialization IDs are definitions, not physical SIDs. ## Prometheus runbook diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 41dd15050..5b4356756 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -203,8 +203,12 @@ struct CompileAndPublishPhysicalPlanRequest { target_collector_ids: Vec, capability_snapshot_id: String, #[serde(default)] + data_snapshot_id: Option, + #[serde(default)] evidence: HashMap, #[serde(default)] + accuracy_evidence: HashMap, + #[serde(default)] exact_composition_costs: HashMap>, #[serde(default)] @@ -588,7 +592,7 @@ fn compile_physical_plan_request( legacy_query_source: planner_types::pre_asap::Source::TimeSeries { metric: query.metric, }, - query_lookback_seconds: query.window_secs, + query_lookback_ms: query.window_secs.saturating_mul(1_000), group_by_labels: query.group_by, accuracy_target: query.accuracy, summary_lifecycle_inputs: query.lifecycle, @@ -597,16 +601,54 @@ fn compile_physical_plan_request( }); } - let planner_selection_trace = match physical::compiler::select_logical_roots_with_trace( - &mut queries, - canonical_roots.clone(), - &request.evidence, - &request.exact_composition_costs, - request.erp.as_ref(), - ) { - Ok(trace) => trace, - Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into())), - }; + let scoped_snapshot_id = request.data_snapshot_id.as_deref().or_else(|| { + request + .workload_cost_evidence + .as_ref() + .map(|evidence| evidence.data_snapshot_id.as_str()) + }); + if request.data_snapshot_id.as_ref().is_some_and(|id| { + request + .workload_cost_evidence + .as_ref() + .is_some_and(|evidence| evidence.data_snapshot_id != *id) + }) { + return Err(( + StatusCode::UNPROCESSABLE_ENTITY, + "accuracy evidence data snapshot differs from workload cost evidence".into(), + )); + } + for (query_id, evidence) in &request.accuracy_evidence { + let Some(query) = queries.iter().find(|query| &query.query_id == query_id) else { + return Err(( + StatusCode::UNPROCESSABLE_ENTITY, + format!("accuracy evidence names unknown query {query_id}").into(), + )); + }; + evidence + .validate( + query_id, + &query.query_string, + &request.data_workload, + scoped_snapshot_id, + now, + request.max_evidence_age_ms, + ) + .map_err(|error| (StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into()))?; + } + let planner_selection_trace = + match physical::compiler::select_logical_roots_with_scoped_evidence_and_trace( + &mut queries, + canonical_roots.clone(), + &request.evidence, + &request.accuracy_evidence, + &request.exact_composition_costs, + request.erp.as_ref(), + now, + ) { + Ok(trace) => trace, + Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into())), + }; for (query, model) in queries.iter_mut().zip(window_models) { physical::compiler::prepare_window_implementations(query, &model, request.target, 0) @@ -620,6 +662,7 @@ fn compile_physical_plan_request( queries, allow_mixed_summary_and_exact_execution: request.target == physical::compiler::PhysicalDeploymentTarget::BackendLocalRemoteWrite, + require_backend_local_execution: false, enabled_materialization_keys: None, topk_membership_evidence_by_query_id: request.evidence, exact_composition_costs: request.exact_composition_costs, @@ -685,7 +728,7 @@ fn compile_physical_plan_request( ) } }, - None => frontend.compile(compilation_request, environment), + None => physical::workload_cost::select_candidates(candidates, environment, None, frontend), }; let bundle = match compiled { Ok(bundle) => bundle, @@ -926,7 +969,7 @@ mod api_tests { "target": "backend_local_remote_write", "queries": [{ "query_id": query.query_id, "query_string": query.query_string, - "metric": metric, "window_secs": query.query_lookback_seconds, "accuracy": query.accuracy_target, + "metric": metric, "window_secs": query.query_lookback_ms / 1_000, "accuracy": query.accuracy_target, "lifecycle": query.summary_lifecycle_inputs, "evaluation_phase_ms": 0, "window_cost_model": { "implementation_id": "test", "cost": query.window_realization_candidates[0].cost } }], "collector_ids": [], "capability_snapshot_id": "test", @@ -938,6 +981,15 @@ mod api_tests { let (plan, collectors, _, _, _) = compile_physical_plan_request(request, false, QueryFrontend::PromQl).unwrap(); let plan = plan.unwrap(); + let report = plan + .cost_comparison + .as_ref() + .expect("HTTP must compare automatically priced candidates"); + assert_eq!(report.model_version, "backend-workload-resources-v1"); + assert!(report + .candidate_evaluations + .iter() + .any(|candidate| candidate.automatic_cost.is_some())); assert!(collectors.is_empty()); assert!(plan.collector_plans.is_empty()); assert_eq!( diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 08ef3bd09..149a03a0b 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -71,7 +71,7 @@ pub struct QueryCompilationInput { /// materialization sources are derived from each post-ASAP SummaryAgg; /// it also remains part of the cost manifest workload identity. pub legacy_query_source: Source, - pub query_lookback_seconds: u64, + pub query_lookback_ms: u64, /// Label names are deployment metadata because Planner's canonical IR /// currently carries positional column IDs at this boundary. pub group_by_labels: Vec, @@ -158,6 +158,8 @@ pub struct PhysicalCompilationRequest { pub planner_selection_trace: Vec, /// Enable a composable DAG with SummaryStore materializations and Prometheus exact subtrees. pub allow_mixed_summary_and_exact_execution: bool, + /// Deployment feasibility: external exact dependencies cannot be bound. + pub require_backend_local_execution: bool, /// Enabled optional candidate keys: None enables all eligible keys; an /// explicitly empty set enables none. These are not catalog definition IDs. pub enabled_materialization_keys: Option>, @@ -196,6 +198,134 @@ pub struct TopKMembershipEvidence { pub source: String, } +/// A proof for one exact Planner operand, including the enforced source +/// domain rather than an observed sample minimum or maximum. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct QuantileOperandDomainEvidence { + pub operand: Value, + pub lower: f64, + pub upper: f64, + pub max_samples: u64, + pub contract: String, +} + +/// Optional accuracy facts bound to one registered query and data snapshot. +/// Missing fields stay unknown to Planner. The data workload and snapshot +/// identity prevent reusing a proof for a different source population. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct ScopedAccuracyEvidence { + pub query_string: String, + pub data_snapshot_id: String, + pub data_workload: DataWorkload, + pub source: String, + pub observed_at_unix_ms: u64, + pub valid_for_ms: u64, + #[serde(default)] + pub quantile_operand_domains: Vec, + #[serde(default)] + pub hll: Option, + #[serde(default)] + pub values_non_negative: Option, + #[serde(default)] + pub input_row_count: Option, + #[serde(default)] + pub hydra_shared_grid_collision_bound: Option, + #[serde(default)] + pub hydra_shared_grid_failure_probability: Option, + #[serde(default)] + pub topk_selected_lower_bound: Option, + #[serde(default)] + pub topk_excluded_upper_bound: Option, + #[serde(default)] + pub topk_interval_failure_probability: Option, +} + +impl ScopedAccuracyEvidence { + pub fn validate( + &self, + query_id: &str, + query_string: &str, + data: &DataWorkload, + snapshot_id: Option<&str>, + now_ms: u64, + max_age_ms: u64, + ) -> Result<(), CompileError> { + let valid_scope = self.query_string == query_string + && &self.data_workload == data + && snapshot_id.is_some_and(|id| id == self.data_snapshot_id) + && !self.data_snapshot_id.trim().is_empty() + && !self.source.trim().is_empty(); + let valid_time = self.observed_at_unix_ms <= now_ms + && self.valid_for_ms > 0 + && now_ms - self.observed_at_unix_ms <= self.valid_for_ms.min(max_age_ms); + let topk_bounds = ( + self.topk_selected_lower_bound, + self.topk_excluded_upper_bound, + self.topk_interval_failure_probability, + ); + let valid_topk = match topk_bounds { + (None, None, None) => true, + (Some(lower), Some(upper), Some(failure)) => { + lower.is_finite() + && upper.is_finite() + && lower > upper + && (0.0..=1.0).contains(&failure) + } + _ => false, + }; + let valid_stats = valid_topk + && self + .hll + .as_ref() + .is_none_or(super::erp::HllConfidenceContract::valid) + && self.input_row_count != Some(0) + && self + .hydra_shared_grid_collision_bound + .is_none_or(|v| v.is_finite() && v >= 0.0) + && self + .hydra_shared_grid_failure_probability + .is_none_or(|v| (0.0..=1.0).contains(&v)) + && self + .quantile_operand_domains + .iter() + .enumerate() + .all(|(index, domain)| { + domain.lower.is_finite() + && domain.upper.is_finite() + && domain.lower <= domain.upper + && (1..=(1u64 << 53)).contains(&domain.max_samples) + && !domain.contract.trim().is_empty() + && !self.quantile_operand_domains[..index] + .iter() + .any(|earlier| earlier.operand == domain.operand) + }); + if valid_scope && valid_time && valid_stats { + return Ok(()); + } + let reason = if self.query_string != query_string { + "accuracy evidence belongs to a different query" + } else if &self.data_workload != data { + "accuracy evidence belongs to a different data workload" + } else if !snapshot_id.is_some_and(|id| id == self.data_snapshot_id) + || self.data_snapshot_id.trim().is_empty() + { + "accuracy evidence belongs to a different or unspecified data snapshot" + } else if self.source.trim().is_empty() { + "accuracy evidence has no provenance source" + } else if !valid_time { + "accuracy evidence is expired, future-dated, or has no validity window" + } else { + "accuracy evidence contains invalid or ambiguous bounds" + }; + Err(CompileError::InvalidEvidence { + query_id: query_id.into(), + reason: reason.into(), + }) + } +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct PhysicalDeploymentContext { @@ -219,7 +349,7 @@ pub enum PhysicalDeploymentTarget { } /// Startup and candidate-discovery input for backend-local planning. -/// Version 2 is the sole supported schema; deployment always requires quotes. +/// Version 2 is the sole supported schema; deployment compares complete workload costs. /// 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. @@ -228,7 +358,7 @@ pub enum PhysicalDeploymentTarget { pub struct BackendLocalPlanningInput { #[serde(rename = "snapshot_version")] pub schema_version: u32, - /// May be absent during candidate discovery, never during deployment. + /// Optional complete provider override; absent evidence uses backend ERP/analytical workload costing. #[serde(default, skip_serializing_if = "Option::is_none")] pub workload_cost_evidence: Option, #[serde(deserialize_with = "deserialize_snapshot_query_workload")] @@ -242,6 +372,9 @@ pub struct BackendLocalPlanningInput { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] pub struct BackendLocalPhysicalInputs { + /// Constrain selection to local execution when no exact upstream is available. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub require_backend_local_execution: bool, pub lifecycle_costs: LifecycleUnitCosts, pub evidence_observed_at_unix_ms: u64, pub evidence_valid_for_ms: u64, @@ -266,6 +399,12 @@ pub struct BackendLocalPhysicalInputs { /// before workload selection so one query cannot borrow another's evidence. #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub topk_evidence: HashMap, + /// Data generation used by scoped accuracy certificates during candidate + /// discovery, before complete workload quotes are available. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub data_snapshot_id: Option, + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub accuracy_evidence: HashMap, /// Measured exact/summary composition profiles keyed by registered /// PromQL. Missing rows keep the corresponding Planner site opaque. #[serde(default, skip_serializing_if = "HashMap::is_empty")] @@ -505,24 +644,75 @@ pub enum CompileError { QueryPlan(#[from] crate::query_plan::QueryPlanError), } -struct QueryEvidence<'a>(Option<&'a TopKMembershipEvidence>); +struct QueryEvidence<'a> { + topk: Option<&'a TopKMembershipEvidence>, + scoped: Option<&'a ScopedAccuracyEvidence>, + now_ms: u64, +} impl AccuracyEvidenceProvider for QueryEvidence<'_> { + fn quantile_input_domain( + &self, + operand: &QueryExpr, + ) -> Option { + let operand = serde_json::to_value(operand).ok()?; + let domain = self + .scoped? + .quantile_operand_domains + .iter() + .find(|entry| entry.operand == operand)?; + Some(asap_aware_mapping::accuracy::QuantileInputDomain { + lower: domain.lower, + upper: domain.upper, + max_samples: domain.max_samples, + contract: domain.contract.clone(), + }) + } + fn propagation_stats( &self, op: &CompositionOperator, _family: &SummaryFamilyType, _query: Option<&SketchQuery>, ) -> PropagationStats { - match (op, self.0) { - (CompositionOperator::TopKSelection, Some(e)) => PropagationStats { - topk_selected_lower_bound: Some(e.selected_lower_bound), - topk_excluded_upper_bound: Some(e.excluded_upper_bound), - topk_interval_failure_probability: Some(e.interval_failure_probability), + let mut stats = self + .scoped + .map_or_else(PropagationStats::default, |evidence| PropagationStats { + values_non_negative: evidence.values_non_negative, + input_row_count: evidence.input_row_count.or_else(|| { + evidence + .data_workload + .input_cardinality + .value_at(self.now_ms) + .copied() + }), + data_distribution: evidence + .data_workload + .distribution + .value_at(self.now_ms) + .cloned(), + hydra_shared_grid_collision_bound: evidence.hydra_shared_grid_collision_bound, + hydra_shared_grid_failure_probability: evidence + .hydra_shared_grid_failure_probability, ..Default::default() - }, - _ => PropagationStats::default(), + }); + if matches!(op, CompositionOperator::TopKSelection) { + if let Some(evidence) = self + .scoped + .filter(|e| e.topk_selected_lower_bound.is_some()) + { + stats.topk_selected_lower_bound = evidence.topk_selected_lower_bound; + stats.topk_excluded_upper_bound = evidence.topk_excluded_upper_bound; + stats.topk_interval_failure_probability = + evidence.topk_interval_failure_probability; + } else if let Some(evidence) = self.topk { + stats.topk_selected_lower_bound = Some(evidence.selected_lower_bound); + stats.topk_excluded_upper_bound = Some(evidence.excluded_upper_bound); + stats.topk_interval_failure_probability = + Some(evidence.interval_failure_probability); + } } + stats } } @@ -546,23 +736,16 @@ impl BackendLocalPlanningInput { self, frontend: QueryFrontend, ) -> Result { - 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 evidence = self.workload_cost_evidence.clone(); let (request, environment) = self.into_physical_compilation_request()?; let candidates = super::workload_cost::enumerate_exact_and_materialized_candidates(request)?; - if frontend == QueryFrontend::MetricsQl { - super::workload_cost::select_lowest_cost_metricsql_candidate( - candidates, - environment, - &evidence, - ) - } else { - super::workload_cost::select_lowest_cost_candidate(candidates, environment, &evidence) - } + super::workload_cost::select_candidates( + candidates, + environment, + evidence.as_ref(), + frontend, + ) } /// Build Planner-authorized candidates for evidence collection without publishing. @@ -582,6 +765,29 @@ impl BackendLocalPlanningInput { } let workload = self.query_workload; let mut data_workload = self.data_workload.clone(); + let scoped_snapshot_id = self + .physical_inputs + .data_snapshot_id + .as_deref() + .or_else(|| { + self.workload_cost_evidence + .as_ref() + .map(|evidence| evidence.data_snapshot_id.as_str()) + }); + if self + .physical_inputs + .data_snapshot_id + .as_ref() + .is_some_and(|id| { + self.workload_cost_evidence + .as_ref() + .is_some_and(|evidence| evidence.data_snapshot_id != *id) + }) + { + return Err(CompileError::Snapshot( + "accuracy evidence data snapshot differs from workload cost evidence".into(), + )); + } if data_workload.data_ingestion_interval.value.is_none() && self.physical_inputs.scrape_interval_ms > 0 { @@ -627,6 +833,7 @@ impl BackendLocalPlanningInput { let mut queries = Vec::with_capacity(entries.len()); let mut canonical_roots = Vec::with_capacity(entries.len()); let mut topk_evidence_by_id = HashMap::new(); + let mut scoped_evidence_by_id = HashMap::new(); for (index, (entry, parsed)) in entries.into_iter().zip(canonical_queries).enumerate() { let evaluation_interval_ms = match entry.recurrence { QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { @@ -638,14 +845,9 @@ impl BackendLocalPlanningInput { ))) } }; - if self.physical_inputs.scrape_interval_ms == 0 - || !self - .physical_inputs - .scrape_interval_ms - .is_multiple_of(1_000) - { + if self.physical_inputs.scrape_interval_ms == 0 { return Err(CompileError::Snapshot( - "scrape_interval_ms must be a positive whole number of seconds".into(), + "scrape_interval_ms must be positive".into(), )); } let accuracy = entry.requirements.accuracy.target(); @@ -673,6 +875,17 @@ impl BackendLocalPlanningInput { if let Some(evidence) = self.physical_inputs.topk_evidence.get(&query_string) { topk_evidence_by_id.insert(query_id.clone(), evidence.clone()); } + if let Some(evidence) = self.physical_inputs.accuracy_evidence.get(&query_string) { + evidence.validate( + &query_id, + &query_string, + &data_workload, + scoped_snapshot_id, + self.environment.observed_at_unix_ms, + self.environment.max_evidence_age_ms, + )?; + scoped_evidence_by_id.insert(query_id.clone(), evidence.clone()); + } queries.push(QueryCompilationInput { query_id, query_string: query_string.clone(), @@ -680,7 +893,7 @@ impl BackendLocalPlanningInput { legacy_query_source: Source::TimeSeries { metric: source_hint, }, - query_lookback_seconds: lookback_ms / 1_000, + query_lookback_ms: lookback_ms, group_by_labels: metadata.group_by_labels, accuracy_target: accuracy, summary_lifecycle_inputs: lifecycle, @@ -688,6 +901,16 @@ impl BackendLocalPlanningInput { materialization_runtime_policy: RuntimeRulePolicy::default(), }); } + if self + .physical_inputs + .accuracy_evidence + .keys() + .any(|query| !queries.iter().any(|entry| &entry.query_string == query)) + { + return Err(CompileError::Snapshot( + "accuracy evidence names a query absent from this workload".into(), + )); + } let mut exact_costs_by_id = HashMap::new(); for (index, entry) in workload.entries().enumerate() { if let Some(rows) = self @@ -709,12 +932,14 @@ impl BackendLocalPlanningInput { exact_costs_by_id.insert(format!("compat-query-{index}"), rows.clone()); } } - let planner_selection_trace = select_logical_roots_with_trace( + let planner_selection_trace = select_logical_roots_with_scoped_evidence_and_trace( &mut queries, canonical_roots.clone(), &topk_evidence_by_id, + &scoped_evidence_by_id, &exact_costs_by_id, self.physical_inputs.erp.as_ref(), + self.environment.observed_at_unix_ms, )?; for query in &mut queries { prepare_window_implementations( @@ -729,6 +954,9 @@ impl BackendLocalPlanningInput { PhysicalCompilationRequest { planner_selection_trace, allow_mixed_summary_and_exact_execution: true, + require_backend_local_execution: self + .physical_inputs + .require_backend_local_execution, enabled_materialization_keys: None, query_workload: Some(workload), data_workload: Some(data_workload), @@ -1083,7 +1311,7 @@ impl DeploymentPlanCompiler { validate_evidence(&query.query_id, e, &environment)?; } let node = query.selected_plan_root.clone(); - reject_uncertified_readouts(&query.query_id, &node)?; + reject_uncertified_readouts(&query.query_id, &node, environment.target)?; let selected = collect_selected_materializations( &node, request.allow_mixed_summary_and_exact_execution, @@ -1092,6 +1320,9 @@ impl DeploymentPlanCompiler { query_id: query.query_id.clone(), reason, })?; + if !selected.is_empty() { + reject_uncertified_readouts(&query.query_id, &node, environment.target)?; + } let selected = selected .into_iter() .filter(|state| { @@ -1210,8 +1441,10 @@ impl DeploymentPlanCompiler { let cohort_nodes = windows::cohort_nodes(&selected); for (ordinal, selected) in selected.into_iter().enumerate() { let mut branch_query = query.clone(); - branch_query.query_lookback_seconds = - selected.window_secs.unwrap_or(query.query_lookback_seconds); + branch_query.query_lookback_ms = selected + .window_secs + .map(|seconds| seconds.saturating_mul(1_000)) + .unwrap_or(query.query_lookback_ms); branch_query.group_by_labels = selected .group_by .clone() @@ -1219,7 +1452,8 @@ impl DeploymentPlanCompiler { branch_query .window_realization_candidates .retain(|candidate| { - candidate.window_secs == branch_query.query_lookback_seconds + candidate.window_secs.saturating_mul(1_000) + == branch_query.query_lookback_ms && if cohort_nodes.contains(&(Rc::as_ptr(&selected.node) as usize)) { windows::is_full_cohort(candidate) } else { @@ -1306,7 +1540,7 @@ impl DeploymentPlanCompiler { let cadence = u64::from( query.summary_lifecycle_inputs.evaluation_interval_ms, ); - let window = query.query_lookback_seconds.saturating_mul(1_000); + let window = query.query_lookback_ms; a % cadence == b % cadence && a % window == b % window } _ => index == query_index, @@ -1697,7 +1931,7 @@ impl DeploymentPlanCompiler { let window_ms = materialization.window_size.saturating_mul(1_000); if materialization_family != *node_family || window_ms == 0 - || source_window.unwrap_or(query.query_lookback_seconds).saturating_mul(1_000) + || source_window.map(|seconds| seconds.saturating_mul(1_000)).unwrap_or(query.query_lookback_ms) % window_ms != 0 { return Err(crate::query_plan::QueryPlanError::Invalid(format!( @@ -1720,7 +1954,7 @@ impl DeploymentPlanCompiler { }) }; let instant = InstantExecution { - lookback_ms: query.query_lookback_seconds.saturating_mul(1_000), + lookback_ms: query.query_lookback_ms, full_history: false, cumulative_readout: true, }; @@ -2026,6 +2260,20 @@ impl DeploymentPlanCompiler { reason: error.to_string(), })?; } + if request.require_backend_local_execution { + for entry in query_plan.entries.values() { + if entry.nodes.values().any(|node| matches!(node, + crate::query_plan::QueryPlanNode::ExactFallback { .. } + | crate::query_plan::QueryPlanNode::Logical { + operator: crate::query_plan::residual::ResidualQueryOperator::ExactSubquery { .. } + | crate::query_plan::residual::ResidualQueryOperator::CandidateExactSubquery { .. }, .. })) { + return Err(CompileError::Query { + query_id: entry.query_id.clone(), + reason: "external execution is unavailable in this deployment".into(), + }); + } + } + } query_plan.validate_against_catalog(&summary_catalog)?; let storage_routing = crate::emit::backend_wire::storage_routing_document( crate::emit::backend_wire::DEFAULT_TENANT, @@ -2101,7 +2349,7 @@ pub fn select_logical_roots_for_queries( select_logical_roots_with_error_resource_profiles(queries, roots, evidence, exact_costs, None) } -fn observed_population_matches_root( +pub(crate) fn observed_population_matches_root( policy: &super::erp::ErpPlanningInput, root: &QueryExpr, ) -> bool { @@ -2164,6 +2412,26 @@ pub fn select_logical_roots_with_trace( evidence: &HashMap, exact_costs: &HashMap>, erp: Option<&super::erp::ErpPlanningInput>, +) -> Result, CompileError> { + select_logical_roots_with_scoped_evidence_and_trace( + queries, + roots, + evidence, + &HashMap::new(), + exact_costs, + erp, + 0, + ) +} + +pub fn select_logical_roots_with_scoped_evidence_and_trace( + queries: &mut [QueryCompilationInput], + roots: Vec>, + evidence: &HashMap, + scoped_evidence: &HashMap, + exact_costs: &HashMap>, + erp: Option<&super::erp::ErpPlanningInput>, + now_ms: u64, ) -> Result, CompileError> { let mut traces = Vec::new(); if roots.len() != queries.len() { @@ -2177,6 +2445,7 @@ pub fn select_logical_roots_with_trace( for (index, root) in roots.into_iter().enumerate() { let accuracy = &queries[index].accuracy_target; let certificate_scope = (evidence.contains_key(&queries[index].query_id) + || scoped_evidence.contains_key(&queries[index].query_id) || exact_costs.contains_key(&queries[index].query_id)) .then(|| queries[index].query_id.clone()); if let Some((_, _, roots)) = cohorts @@ -2189,13 +2458,8 @@ pub fn select_logical_roots_with_trace( } } for (accuracy, scope, roots) in cohorts { - // ERP v1 has no calibrated failure probability. Preserve explicit - // confidence requirements through theoretical/exact fallback. let scoped_erp = erp.map(|policy| { let mut policy = policy.clone(); - if !matches!(accuracy, AccuracyTarget::Epsilon(_)) { - policy.artifact.records.clear(); - } if policy.observed_populations.is_some() && !roots .iter() @@ -2218,7 +2482,15 @@ pub fn select_logical_roots_with_trace( }); policy }); - let erp = scoped_erp.as_ref(); + // Confidence limitations invalidate an empirical accuracy decision, + // not the independently matched resource measurements. + let accuracy_erp = scoped_erp.clone().map(|mut policy| { + if !matches!(accuracy, AccuracyTarget::Epsilon(_)) { + policy.artifact.records.clear(); + } + policy + }); + let erp = accuracy_erp.as_ref(); let mut model = ControlPlaneCostModel::new(accuracy.clone()).with_exact_composition_costs( scope .as_ref() @@ -2229,8 +2501,17 @@ pub fn select_logical_roots_with_trace( if let Some(erp) = erp { model = model.with_erp(erp.clone()); } + if let Some(costs) = &scoped_erp { + model = model.with_erp_costs(costs.clone()); + } let certificate = scope.as_ref().and_then(|id| evidence.get(id)); + let scoped_certificate = scope.as_ref().and_then(|id| scoped_evidence.get(id)); + let hll = scoped_certificate.and_then(|evidence| evidence.hll.as_ref()); + if let Some(contract) = hll { + model = model.with_hll_confidence(contract.clone()); + } let accuracy_model = super::erp::ErpAccuracyModel { + hll, policy: erp, max_error: match accuracy { AccuracyTarget::Epsilon(e) | AccuracyTarget::EpsilonDelta { epsilon: e, .. } => e, @@ -2242,10 +2523,23 @@ pub fn select_logical_roots_with_trace( roots, accuracy, &model, - &QueryEvidence(certificate), + &QueryEvidence { + topk: certificate, + scoped: scoped_certificate, + now_ms, + }, &accuracy_model, ) .map_err(|error| CompileError::Snapshot(error.to_string()))?; + if let Some(evidence) = scoped_certificate { + trace["accuracy_evidence_scope"] = serde_json::json!({ + "query_id": scope, + "source": evidence.source, + "data_snapshot_id": evidence.data_snapshot_id, + "observed_at_unix_ms": evidence.observed_at_unix_ms, + "hll": evidence.hll, + }); + } trace["deployment_overrides"] = serde_json::json!([]); let selected_indices = selected.iter().map(|(index, _)| *index).collect::>(); for (index, node) in selected { @@ -2425,7 +2719,11 @@ pub fn select_post_asap( expr, &CollectorFixtureModel(model), &DefaultAccuracyModel, - &QueryEvidence(evidence), + &QueryEvidence { + topk: evidence, + scoped: None, + now_ms: 0, + }, ) } @@ -2585,10 +2883,9 @@ pub(super) fn derived_window_cost( /// example, `a[1m] offset 1h` selects `(t - 61m, t - 60m]`. fn query_history_window_ms(expr: &QueryExpr, scrape_interval_ms: u64) -> Result { fn duration_ms(duration: std::time::Duration) -> Result { - if duration.subsec_nanos() != 0 { + if !duration.subsec_nanos().is_multiple_of(1_000_000) { return Err(CompileError::Snapshot( - "PromQL ranges must be a whole number of seconds in the backend-local profile" - .into(), + "PromQL ranges must have millisecond precision".into(), )); } u64::try_from(duration.as_millis()) @@ -2617,17 +2914,10 @@ fn query_history_window_ms(expr: &QueryExpr, scrape_interval_ms: u64) -> Result< duration_ms(*range)?, visit(child, scrape_interval_ms, true)?, ), - QueryExpr::TimeShift { shift, child } => { - if !shift.offset_ms.unsigned_abs().is_multiple_of(1_000) { - return Err(CompileError::Snapshot( - "PromQL offsets must be a whole number of seconds in the backend-local profile".into(), - )); - } - add( - shift.offset_ms.max(0) as u64, - visit(child, scrape_interval_ms, in_range)?, - ) - } + QueryExpr::TimeShift { shift, child } => add( + shift.offset_ms.max(0) as u64, + visit(child, scrape_interval_ms, in_range)?, + ), QueryExpr::PromqlScalarBridge(child) | QueryExpr::PromqlVectorFromScalar(child) | QueryExpr::PromqlScalarFromVector(child) @@ -2773,7 +3063,7 @@ pub(super) fn validate_window_implementations( && evidence.cpu_cost >= 0.0 && evidence.weighted_cost.is_finite() && evidence.weighted_cost >= 0.0 - && candidate.window_secs == query.query_lookback_seconds + && candidate.window_secs.saturating_mul(1_000) == query.query_lookback_ms && candidate .layout .validate(candidate.window_secs, candidate.slide_secs) @@ -2836,8 +3126,8 @@ fn retained_state_bytes(materialization: &asap_types::PrecomputeMaterialization) ) } -pub(super) fn estimated_state_bytes( - aggregation: &asap_types::AggregationType, +pub(crate) fn estimated_state_bytes( + aggregation_type: &asap_types::AggregationType, parameters: &HashMap, ) -> u128 { use asap_types::AggregationType as A; @@ -2848,7 +3138,7 @@ pub(super) fn estimated_state_bytes( .find_map(|name| parameters.get(*name).and_then(Value::as_u64)) .unwrap_or(fallback) as u128 }; - match aggregation { + match aggregation_type { A::CountMinSketch | A::CountSketch => { parameter(&["width", "w", "col_num", "col"], 1) * parameter(&["depth", "d", "row_num", "row"], 1) @@ -2880,7 +3170,7 @@ pub(super) fn estimated_state_bytes( } } -fn retained_partition_count( +pub(crate) fn retained_partition_count( materialization: &asap_types::PrecomputeMaterialization, input_cardinality: Option, ) -> u128 { @@ -2985,8 +3275,8 @@ fn select_lifecycle( raw_materialization_input_contract(node) .ok() .and_then(|(_, window, _)| window) - .unwrap_or(query.query_lookback_seconds) - .saturating_mul(1_000), + .map(|seconds| seconds.saturating_mul(1_000)) + .unwrap_or(query.query_lookback_ms), )), as_of: None, }, @@ -3301,7 +3591,11 @@ fn validate_executable_subdag(node: &Rc) -> Result<(), String> { Ok(()) } -fn reject_uncertified_readouts(query_id: &str, root: &Rc) -> Result<(), CompileError> { +fn reject_uncertified_readouts( + query_id: &str, + root: &Rc, + target: PhysicalDeploymentTarget, +) -> Result<(), CompileError> { let dag = planner_types::post_asap::compile_executable_dag(root).map_err(|error| { CompileError::Query { query_id: query_id.into(), @@ -3309,6 +3603,21 @@ fn reject_uncertified_readouts(query_id: &str, root: &Rc) -> Result } })?; for node in &dag.nodes { + if target != PhysicalDeploymentTarget::BackendLocalRemoteWrite + && node.guarantee.as_ref().is_some_and(|guarantee| { + guarantee.provenance.iter().any(|source| { + matches!(source, + planner_types::post_asap::GuaranteeSource::SketchReadout { contract, .. } + if contract == "classic_hll_linear_counting_collision_bound_v1") + }) + }) + { + return Err(CompileError::Query { + query_id: query_id.into(), + reason: "classic HLL confidence is bound to the backend-local Regular estimator" + .into(), + }); + } if matches!( node.payload, planner_types::post_asap::ExecutableOperatorPayload::SummaryEstimate { .. } @@ -3336,7 +3645,9 @@ fn physical_aggregation( aggregation_id, metric_name: selected.metric.clone(), family: selected.family.clone(), - window_secs: selected.window_secs.unwrap_or(query.query_lookback_seconds), + window_secs: selected + .window_secs + .unwrap_or(query.query_lookback_ms.div_ceil(1_000)), spatial_filter: selected.spatial_filter.clone(), grouping: selected .group_by @@ -3731,7 +4042,7 @@ fn collect_selected_materializations( Ok(selected) } -pub(super) fn sketch_params_json(params: &planner_types::post_asap::SketchParams) -> Value { +pub(crate) fn sketch_params_json(params: &planner_types::post_asap::SketchParams) -> Value { use planner_types::post_asap::SketchParams as P; match params { P::UnivMon { @@ -3830,6 +4141,7 @@ pub(crate) mod tests { "quantile by (job) (0.99, a)", "topk by (job) (1, a)", "topk by (job) (5, a)", + "count by (job) (a)", ]; snapshot.query_workload.repeating_queries = Some( queries @@ -4652,6 +4964,7 @@ pub(crate) mod tests { planner_selection_trace: Vec::new(), canonical_roots: Vec::new(), allow_mixed_summary_and_exact_execution: false, + require_backend_local_execution: false, enabled_materialization_keys: None, query_workload: None, data_workload: None, @@ -4660,7 +4973,7 @@ pub(crate) mod tests { query_string: promql.into(), selected_plan_root: post_asap, legacy_query_source: Source::TimeSeries { metric: "m".into() }, - query_lookback_seconds: 60, + query_lookback_ms: 60_000, group_by_labels: vec![], accuracy_target: accuracy, summary_lifecycle_inputs: lifecycle, @@ -4913,6 +5226,273 @@ pub(crate) mod tests { assert!(result.unwrap().precompute_plan.materializations.is_empty()); } + fn bounded_hll_snapshot_wire() -> serde_json::Value { + let mut wire = serde_json::to_value(planning_snapshot()).unwrap(); + let query = "distinct_over_time(data[5s])"; + wire["query_workload"]["repeating_queries"][0]["query"] = query.into(); + wire["query_workload"]["repeating_queries"][0]["requirements"]["accuracy"] = + serde_json::json!({"explicit":{"EpsilonDelta":{"epsilon":0.05,"delta":0.01}}}); + wire["implementation"]["data_snapshot_id"] = "hll-bounded-population".into(); + let now = wire["environment"]["observed_at_unix_ms"].clone(); + wire["implementation"]["accuracy_evidence"] = serde_json::json!({query: { + "query_string": query, "data_snapshot_id": "hll-bounded-population", + "data_workload": wire["data_workload"], "source": "enforced-distinct-domain-v1", + "observed_at_unix_ms": now, "valid_for_ms": 60000, + "hll": {"model":"asap-classic64-uniform-hash-linear-counting-v1", + "max_distinct_per_readout": 128} + }}); + wire + } + + /// An applicable classic-HLL confidence contract restores normal selection. + #[test] + fn bounded_hll_confidence_selects_and_binds_a_materialization() { + let wire = bounded_hll_snapshot_wire(); + let snapshot: BackendLocalPlanningInput = serde_json::from_value(wire).unwrap(); + let (request, environment) = snapshot.into_physical_compilation_request().unwrap(); + let plan = DeploymentPlanCompiler + .compile_promql(request, environment) + .unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 1, "{plan:#?}"); + assert_eq!( + plan.precompute_plan.materializations[0].aggregation_type, + asap_types::AggregationType::HLL + ); + } + + /// A backend-local estimator proof cannot certify an unverified collector implementation. + #[test] + fn hll_confidence_cannot_be_rebound_to_collectors() { + let snapshot: BackendLocalPlanningInput = + serde_json::from_value(bounded_hll_snapshot_wire()).unwrap(); + let (request, _) = snapshot.into_physical_compilation_request().unwrap(); + let error = reject_uncertified_readouts( + "q", + &request.queries[0].selected_plan_root, + PhysicalDeploymentTarget::DistributedCollectors, + ) + .unwrap_err(); + assert!(error + .to_string() + .contains("backend-local Regular estimator")); + } + + /// Missing contracts and unattainable targets cannot acquire a confidence proof. + #[test] + fn hll_confidence_keeps_exact_when_absent_or_insufficient() { + for missing in [true, false] { + let mut wire = bounded_hll_snapshot_wire(); + if missing { + wire["implementation"]["accuracy_evidence"] = serde_json::json!({}); + } else { + wire["query_workload"]["repeating_queries"][0]["requirements"]["accuracy"] = + serde_json::json!({"explicit":{"EpsilonDelta":{"epsilon":0.05,"delta":1e-12}}}); + } + let snapshot: BackendLocalPlanningInput = serde_json::from_value(wire).unwrap(); + let (request, environment) = snapshot.into_physical_compilation_request().unwrap(); + let plan = DeploymentPlanCompiler + .compile_promql(request, environment) + .unwrap(); + assert!( + plan.precompute_plan.materializations.is_empty(), + "{plan:#?}" + ); + } + } + + /// An estimator mismatch or unsupported population must be rejected at admission. + #[test] + fn hll_confidence_rejects_invalid_source_contracts() { + for contract in [ + serde_json::json!({"model":"asap-classic64-uniform-hash-linear-counting-v1","max_distinct_per_readout":0}), + serde_json::json!({"model":"asap-classic64-uniform-hash-linear-counting-v1","max_distinct_per_readout":4097}), + serde_json::json!({"model":"hip-rse","max_distinct_per_readout":128}), + ] { + let mut wire = bounded_hll_snapshot_wire(); + wire["implementation"]["accuracy_evidence"]["distinct_over_time(data[5s])"]["hll"] = + contract; + let snapshot: BackendLocalPlanningInput = serde_json::from_value(wire).unwrap(); + assert!(snapshot.into_physical_compilation_request().is_err()); + } + } + + /// Accuracy facts must belong to the same query, workload, snapshot and + /// evidence window before they enter Planner. + #[test] + fn scoped_accuracy_evidence_rejects_wrong_or_stale_facts() { + let snapshot: BackendLocalPlanningInput = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let evidence = ScopedAccuracyEvidence { + query_string: "quantile_over_time(0.9,m[1m])".into(), + data_snapshot_id: "snapshot-a".into(), + data_workload: snapshot.data_workload.clone(), + source: "enforced-source-contract".into(), + observed_at_unix_ms: 9_000, + valid_for_ms: 2_000, + hll: None, + quantile_operand_domains: vec![], + values_non_negative: Some(true), + input_row_count: Some(10), + hydra_shared_grid_collision_bound: None, + hydra_shared_grid_failure_probability: None, + topk_selected_lower_bound: None, + topk_excluded_upper_bound: None, + topk_interval_failure_probability: None, + }; + let validate = |evidence: &ScopedAccuracyEvidence, query: &str, snapshot_id: &str, now| { + evidence.validate( + "q", + query, + &snapshot.data_workload, + Some(snapshot_id), + now, + 2_000, + ) + }; + assert!(validate(&evidence, &evidence.query_string, "snapshot-a", 10_000).is_ok()); + assert!(validate(&evidence, "another query", "snapshot-a", 10_000).is_err()); + assert!(validate(&evidence, &evidence.query_string, "snapshot-b", 10_000).is_err()); + assert!(validate(&evidence, &evidence.query_string, "snapshot-a", 12_000).is_err()); + let mut invalid = evidence.clone(); + invalid.hydra_shared_grid_failure_probability = Some(1.5); + assert!(validate(&invalid, &invalid.query_string, "snapshot-a", 10_000).is_err()); + let mut partial_topk = evidence.clone(); + partial_topk.topk_selected_lower_bound = Some(10.0); + assert!(validate( + &partial_topk, + &partial_topk.query_string, + "snapshot-a", + 10_000 + ) + .is_err()); + let mut valid_topk = partial_topk; + valid_topk.topk_excluded_upper_bound = Some(8.0); + valid_topk.topk_interval_failure_probability = Some(0.01); + assert!(validate(&valid_topk, &valid_topk.query_string, "snapshot-a", 10_000).is_ok()); + let stats = QueryEvidence { + topk: None, + scoped: Some(&valid_topk), + now_ms: 10_000, + } + .propagation_stats( + &CompositionOperator::TopKSelection, + &SummaryFamilyType::ExactAggregate( + planner_types::post_asap::ExactKind::Count, + planner_types::post_asap::ExactParams::Count, + ), + None, + ); + assert_eq!(stats.topk_selected_lower_bound, Some(10.0)); + assert_eq!(stats.topk_excluded_upper_bound, Some(8.0)); + } + + /// Quantile domain proof applies only to the operand named by its AST, + /// even when both operands read the same metric. + #[test] + fn scoped_quantile_domain_matches_one_exact_operand() { + let snapshot: BackendLocalPlanningInput = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let accuracy = AccuracyTarget::EpsilonDelta { + epsilon: 0.05, + delta: 0.01, + }; + let root = crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.9,m[5m]) / quantile_over_time(0.5,m[5m])", + accuracy.clone(), + ) + .unwrap(); + let QueryExpr::BinaryOp { lhs, rhs, .. } = &root else { + panic!("expected ratio expression") + }; + let scoped = ScopedAccuracyEvidence { + query_string: "quantile_over_time(0.9,m[5m]) / quantile_over_time(0.5,m[5m])".into(), + data_snapshot_id: "snapshot-a".into(), + data_workload: snapshot.data_workload, + source: "enforced-source-contract".into(), + observed_at_unix_ms: 9_000, + valid_for_ms: 2_000, + hll: None, + quantile_operand_domains: vec![QuantileOperandDomainEvidence { + operand: serde_json::to_value(lhs.as_ref()).unwrap(), + lower: 1.0, + upper: 100.0, + max_samples: 10_000, + contract: "source-schema-v1".into(), + }], + values_non_negative: None, + input_row_count: None, + hydra_shared_grid_collision_bound: None, + hydra_shared_grid_failure_probability: None, + topk_selected_lower_bound: None, + topk_excluded_upper_bound: None, + topk_interval_failure_probability: None, + }; + let provider = QueryEvidence { + topk: None, + scoped: Some(&scoped), + now_ms: 10_000, + }; + assert!(provider.quantile_input_domain(lhs).is_some()); + assert!(provider.quantile_input_domain(rhs).is_none()); + let inspect = |provider: &dyn AccuracyEvidenceProvider| { + let (_, trace) = + crate::planner_selection::select_workload_with_accuracy_model_and_trace( + vec![(0, Rc::new(root.clone()))], + accuracy.clone(), + &ControlPlaneCostModel::new(accuracy.clone()), + provider, + &DefaultAccuracyModel, + ) + .unwrap(); + trace + }; + let unknown = inspect(&provider); + assert!(unknown["groups"].as_array().unwrap().iter().any(|group| { + group["candidates"] + .as_array() + .unwrap() + .iter() + .any(|candidate| { + candidate["accuracy_status"] == "unknown" && candidate["selected"] == false + }) + })); + let mut complete = scoped.clone(); + complete + .quantile_operand_domains + .push(QuantileOperandDomainEvidence { + operand: serde_json::to_value(rhs.as_ref()).unwrap(), + lower: 1.0, + upper: 100.0, + max_samples: 10_000, + contract: "source-schema-v1".into(), + }); + let complete_provider = QueryEvidence { + topk: None, + scoped: Some(&complete), + now_ms: 10_000, + }; + assert!(complete_provider.quantile_input_domain(rhs).is_some()); + let complete_trace = inspect(&complete_provider); + assert!(complete_trace["groups"] + .as_array() + .unwrap() + .iter() + .any(|group| { + group["candidates"] + .as_array() + .unwrap() + .iter() + .any(|candidate| { + candidate["accuracy_status"] == "known" + && candidate["replacement_kind"] == "summary" + }) + })); + } + #[test] fn snapshot_metricsql_entry_uses_the_shared_serving_language_contract() { let snapshot: BackendLocalPlanningInput = serde_json::from_str(include_str!( @@ -6090,6 +6670,29 @@ pub(crate) mod tests { .unwrap() } + /// ERP memory remains usable under an explicit confidence target, even + /// though ERP v1 error observations cannot certify that target. + #[test] + fn erp_resource_costs_survive_confidence_target() { + let mut snapshot = planning_snapshot(); + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query("quantile_over_time(0.9,m[1m])".into()); + entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::EpsilonDelta { + epsilon: 0.1, + delta: 0.01, + }); + snapshot.physical_inputs.erp = + Some(crate::physical::post_asap::cost_model::tests::erp_cost_fixture()); + let (request, _) = snapshot.into_physical_compilation_request().unwrap(); + assert!(request + .planner_selection_trace + .iter() + .flat_map(|trace| trace["groups"].as_array().unwrap()) + .flat_map(|group| group["candidates"].as_array().unwrap()) + .any(|candidate| candidate["cost_estimate"]["source"] == "erp" + && candidate["estimated_cost"] == 7777.0)); + } + fn derived_query_window_secs(query: &str) -> u64 { let mut snapshot = planning_snapshot(); let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; @@ -6100,7 +6703,8 @@ pub(crate) mod tests { .unwrap() .0 .queries[0] - .query_lookback_seconds + .query_lookback_ms + / 1_000 } fn set_data_ingestion_interval( @@ -6125,7 +6729,7 @@ pub(crate) mod tests { set_data_ingestion_interval(&mut snapshot, Some(1_000)); let (request, _) = snapshot.into_physical_compilation_request().unwrap(); - assert_eq!(request.queries[0].query_lookback_seconds, 1); + assert_eq!(request.queries[0].query_lookback_ms, 1_000); } // Snapshot lowering must use the environment clock for cadence freshness. @@ -6165,7 +6769,7 @@ pub(crate) mod tests { request.data_workload.unwrap().data_ingestion_interval.value, Some(DurationMs(5_000)) ); - assert_eq!(request.queries[0].query_lookback_seconds, 5); + assert_eq!(request.queries[0].query_lookback_ms, 5_000); } // An explicitly invalid interval must not be replaced by the migration value. @@ -6200,31 +6804,50 @@ pub(crate) mod tests { assert_eq!(derived_query_window_secs("sum_over_time(a[1s])"), 1); } - // The seconds-based backend must fail explicitly instead of shrinking history. + /// Fractional ranges, subqueries and offsets preserve their exact history. #[test] - fn derived_history_rejects_fractional_seconds() { - for query in [ - "sum_over_time(a[1500ms])", - "sum_over_time(a[500ms])", - "sum_over_time(a[1500ms] offset 500ms)", - "sum_over_time(a[1s]) + sum(sum_over_time(b[500ms]))", - "avg_over_time((sum(a))[1500ms:])", - "sum(a offset 500ms)", + fn derived_history_preserves_milliseconds() { + for (query, expected) in [ + ("sum_over_time(a[1500ms])", 1500), + ("sum_over_time(a[500ms])", 500), + ("sum_over_time(a[1500ms] offset 500ms)", 2000), + ("sum_over_time(a[1s]) + sum(sum_over_time(b[500ms]))", 1000), + ("avg_over_time((sum(a))[1500ms:])", 6500), + ("sum(a offset 500ms)", 5500), ] { let mut snapshot = planning_snapshot(); snapshot.query_workload.repeating_queries.as_mut().unwrap()[0].query = Query(query.into()); - let error = snapshot - .into_physical_compilation_request() - .expect_err(query) - .to_string(); - assert!( - error.contains("whole number of seconds"), - "{query}: {error}" - ); + let (request, _) = snapshot.into_physical_compilation_request().expect(query); + assert_eq!(request.queries[0].query_lookback_ms, expected, "{query}"); } } + /// A real 100 ms source stays at 100 ms through lowering and query publication. + #[test] + fn hundred_millisecond_source_compiles_without_rounding() { + let mut snapshot = planning_snapshot(); + snapshot.query_workload.repeating_queries.as_mut().unwrap()[0].query = + Query("sum(data)".into()); + snapshot.physical_inputs.scrape_interval_ms = 100; + set_data_ingestion_interval(&mut snapshot, Some(100)); + let (request, environment) = snapshot.into_physical_compilation_request().unwrap(); + assert_eq!(request.queries[0].query_lookback_ms, 100); + let plan = DeploymentPlanCompiler + .compile_promql(request, environment) + .unwrap(); + assert_eq!( + plan.query_plan + .entries + .values() + .next() + .unwrap() + .instant + .lookback_ms, + 100 + ); + } + #[test] fn snapshot_rejects_manual_lookback() { let mut wire = serde_json::to_value(planning_snapshot()).unwrap(); @@ -6383,7 +7006,7 @@ pub(crate) mod tests { (60_000, 90_000), ] { let mut query = request.queries[0].clone(); - query.query_lookback_seconds = lookback_ms / 1_000; + query.query_lookback_ms = lookback_ms; let expr = crate::query_parser::parse_query_expr_canonical( &format!("quantile_over_time(0.5, data[{}s])", lookback_ms / 1_000), AccuracyTarget::Exact, @@ -7254,6 +7877,7 @@ pub(crate) mod tests { query_workload, data_workload, physical_inputs: BackendLocalPhysicalInputs { + require_backend_local_execution: false, lifecycle_costs: template.summary_lifecycle_inputs.costs, evidence_observed_at_unix_ms: 9_500, evidence_valid_for_ms: 60_000, @@ -7267,6 +7891,8 @@ pub(crate) mod tests { query_retention_margin_ms: 0, retained_summary_memory_budget_bytes: DEFAULT_RETAINED_SUMMARY_MEMORY_BUDGET_BYTES, topk_evidence: HashMap::new(), + data_snapshot_id: None, + accuracy_evidence: HashMap::new(), exact_composition_costs: HashMap::new(), erp: None, }, @@ -7398,8 +8024,13 @@ pub(crate) mod tests { assert_eq!(encoded, fixture); assert!( - snapshot.clone().compile_promql().is_err(), - "discovery fixtures must be priced before deployment" + snapshot + .clone() + .compile_promql() + .unwrap() + .cost_comparison + .is_some(), + "deployment computes complete workload costs automatically" ); let (local, env) = snapshot .clone() @@ -7438,8 +8069,13 @@ pub(crate) mod tests { let snapshot: BackendLocalPlanningInput = serde_json::from_str(source).expect("strict compatibility demo fixture"); assert!( - snapshot.clone().compile_promql().is_err(), - "discovery fixtures must be priced before deployment" + snapshot + .clone() + .compile_promql() + .unwrap() + .cost_comparison + .is_some(), + "deployment computes complete workload costs automatically" ); let (local, env) = snapshot .clone() @@ -7703,9 +8339,9 @@ pub(crate) mod tests { let mut second = request("q40", "sum(sum_over_time(m[40s]))") .queries .remove(0); - workload.queries[0].query_lookback_seconds = 20; + workload.queries[0].query_lookback_ms = 20_000; workload.queries[0].window_realization_candidates[0].window_secs = 20; - second.query_lookback_seconds = 40; + second.query_lookback_ms = 40_000; second.summary_lifecycle_inputs.evaluation_interval_ms = 20_000; second.window_realization_candidates[0].window_secs = 40; workload.queries.push(second); diff --git a/control_plane/src/physical/compiler/windows.rs b/control_plane/src/physical/compiler/windows.rs index 072ee7c2e..709273322 100644 --- a/control_plane/src/physical/compiler/windows.rs +++ b/control_plane/src/physical/compiler/windows.rs @@ -157,7 +157,9 @@ pub fn prepare_window_implementations( .iter() .map(|state| { ( - state.window_secs.unwrap_or(query.query_lookback_seconds), + state + .window_secs + .unwrap_or(query.query_lookback_ms.div_ceil(1_000)), cohorts.contains(&(Rc::as_ptr(&state.node) as usize)), ) }) diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index 26bf26af2..97d6e9ddd 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -204,10 +204,44 @@ impl ReadoutEvidence { } } -/// ERP v1 measures error magnitudes, not tail probabilities. Only an explicit -/// epsilon-only request may use these observations as its accuracy contract. +/// Trusted source contract for every HLL readout population of one scoped query. +/// The upper bound covers the union of all merged panes, not each pane separately. +/// This is not derived from observed series count or a sampled distinct count. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct HllConfidenceContract { + /// Identifies classic 64-bit HLL with its linear-counting correction, under + /// independent uniform bucket hashing. Other estimators are not certified. + pub model: String, + pub max_distinct_per_readout: u32, +} + +impl HllConfidenceContract { + pub(crate) fn valid(&self) -> bool { + self.model == "asap-classic64-uniform-hash-linear-counting-v1" + && (1..=4096).contains(&self.max_distinct_per_readout) + } + + pub(crate) fn confidence( + &self, + epsilon: f64, + ) -> Option { + self.valid() + .then(|| { + asap_aware_mapping::hll_confidence::ClassicHllConfidence::new( + self.max_distinct_per_readout, + epsilon, + ) + }) + .flatten() + } +} + +/// ERP error magnitudes and estimator-specific confidence are separate inputs. +/// ERP v1 observed maxima do not establish a failure-probability guarantee. pub(crate) struct ErpAccuracyModel<'a> { pub policy: Option<&'a ErpPlanningInput>, + pub hll: Option<&'a HllConfidenceContract>, pub max_error: f64, } @@ -224,6 +258,20 @@ impl asap_aware_mapping::AccuracyModel for ErpAccuracyModel<'_> { query: &planner_types::post_asap::SketchQuery, ) -> Option { use planner_types::post_asap::*; + if let ( + Some(contract), + SummaryFamilyType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance), + SketchQuery::Cardinality, + ) = (self.hll, family, query) + { + if let (SketchAlgorithm::Hll, SketchParams::Hll { precision }) = + (kind.algorithm(), kind.params()) + { + // Resource ERP can still rank this state. Error maxima must not + // replace the independent, estimator-specific confidence bound. + return contract.confidence(self.max_error)?.guarantee(*precision); + } + } let theoretical = asap_aware_mapping::DefaultAccuracyModel.local_guarantee(family, query); let (Some(policy), SummaryFamilyType::Sketch(kind, _)) = (self.policy, family) else { return theoretical; @@ -575,6 +623,91 @@ impl ErpParameterDecision { } impl ErpPlanningInput { + /// Match ERP resource measurements for an already-sized candidate. Error + /// magnitudes only identify existing records here: no accuracy target is + /// certified by this cost lookup. Reuse ERP's implementation, population, + /// shape, minimum-trial and runtime checks rather than a second matcher. + pub(crate) fn candidate_resources( + &self, + algorithm: &SketchAlgorithm, + params: &SketchParams, + ) -> Option<(asap_aware_mapping::erp::ErpResourceProfile, Vec)> { + self.artifact.validate().ok()?; + let metrics = self + .artifact + .records + .iter() + .flat_map(|row| row.error_metrics.keys().cloned()) + .collect::>(); + metrics + .into_iter() + .filter_map(|metric| { + let ErpParameterDecision::Empirical { + record_id, + params: selected, + .. + } = self.select_metric( + algorithm.clone(), + &metric, + f64::MAX, + params.clone(), + Some(params), + ) + else { + return None; + }; + if &selected != params { + return None; + } + let mut ids = if self.artifact.records.iter().any(|row| row.id == record_id) { + vec![record_id] + } else { + serde_json::from_str::>(&record_id).ok()? + }; + if ids.is_empty() { + return None; + } + // The logical proxy is per partition. For population-specific + // profiles use the largest matched partition, not a pooled sketch. + let mut resources = asap_aware_mapping::erp::ErpResourceProfile { + memory_bytes: 0.0, + update_cpu_seconds: 0.0, + merge_cpu_seconds: 0.0, + query_cpu_seconds: 0.0, + }; + for id in &ids { + let row = self.artifact.records.iter().find(|row| &row.id == id)?; + resources.memory_bytes = resources.memory_bytes.max(row.resources.memory_bytes); + resources.update_cpu_seconds = resources + .update_cpu_seconds + .max(row.resources.update_cpu_seconds); + resources.merge_cpu_seconds = resources + .merge_cpu_seconds + .max(row.resources.merge_cpu_seconds); + resources.query_cpu_seconds = resources + .query_cpu_seconds + .max(row.resources.query_cpu_seconds); + } + ids.sort(); + ids.dedup(); + Some((resources, ids)) + }) + .min_by(|a, b| { + a.0.memory_bytes + .total_cmp(&b.0.memory_bytes) + .then_with(|| a.1.cmp(&b.1)) + }) + } + + pub(crate) fn candidate_memory_bytes( + &self, + algorithm: &SketchAlgorithm, + params: &SketchParams, + ) -> Option<(f64, Vec)> { + self.candidate_resources(algorithm, params) + .map(|(resources, ids)| (resources.memory_bytes, ids)) + } + pub fn select( &self, algorithm: SketchAlgorithm, @@ -873,7 +1006,10 @@ fn u32_param(parameters: &Value, names: &[&str]) -> Option { .then_some(value as u32) } -fn parse_params(algorithm: &SketchAlgorithm, parameters: &Value) -> Option { +pub(crate) fn parse_params( + algorithm: &SketchAlgorithm, + parameters: &Value, +) -> Option { let width = || u32_param(parameters, &["width", "cols"]); let depth = || u32_param(parameters, &["depth", "rows"]); Some(match algorithm { @@ -1080,6 +1216,7 @@ mod tests { GroupingStrategy::PerSubpopulationInstance, ); let model = ErpAccuracyModel { + hll: None, policy: Some(&policy), max_error: 0.2, }; @@ -1093,6 +1230,7 @@ mod tests { )); policy.artifact.records[1].error_metrics.clear(); assert!(ErpAccuracyModel { + hll: None, policy: Some(&policy), max_error: 0.2 } @@ -1114,6 +1252,7 @@ mod tests { GroupingStrategy::PerSubpopulationInstance, ); let model = ErpAccuracyModel { + hll: None, policy: Some(&policy), max_error: 0.05, }; @@ -1160,6 +1299,7 @@ mod tests { GroupingStrategy::PerSubpopulationInstance, ); let model = ErpAccuracyModel { + hll: None, policy: Some(&policy), max_error: 0.2, }; @@ -1191,6 +1331,7 @@ mod tests { .error_metrics .remove("max_frequency_entropy_absolute_bits_error"); let model = ErpAccuracyModel { + hll: None, policy: Some(&policy), max_error: 0.2, }; @@ -1444,6 +1585,15 @@ mod tests { assert!(invalid.populations.is_empty()); assert!(invalid.invalid_reason.as_deref().unwrap().contains("stale")); assert!(policy.observed_shape.is_none()); + assert!(policy + .candidate_memory_bytes( + &SketchAlgorithm::Cms, + &SketchParams::Cms { + width: 512, + depth: 3 + } + ) + .is_none()); } /// Only the activated catalog may supply a candidate's data contract. @@ -1604,6 +1754,16 @@ mod tests { minimum_confidence: 0.9, minimum_confidence_margin: 0.05, }); + assert_eq!( + policy.candidate_memory_bytes( + &SketchAlgorithm::Cms, + &SketchParams::Cms { + width: 512, + depth: 3 + } + ), + Some((12_288.0, vec!["cms-512".into()])) + ); assert!(matches!( policy.select( SketchAlgorithm::Cms, @@ -1647,6 +1807,15 @@ mod tests { policy.select(SketchAlgorithm::Cms, 0.01, theory.clone()), ErpParameterDecision::TheoreticalFallback { params, .. } if params == theory )); + assert!(policy + .candidate_memory_bytes( + &SketchAlgorithm::Cms, + &SketchParams::Cms { + width: 512, + depth: 3 + } + ) + .is_none()); } #[test] diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs index 9610ed0a1..8e0a4a3b5 100644 --- a/control_plane/src/physical/post_asap/cost_model.rs +++ b/control_plane/src/physical/post_asap/cost_model.rs @@ -167,6 +167,8 @@ pub struct ControlPlaneCostModel { offline_frequency_comparison: Option<(OfflineComparisonEvidence, OfflineComparisonRequest)>, exact_composition_costs: Vec, erp: Option, + erp_costs: Option, + hll_confidence: Option, } impl ControlPlaneCostModel { @@ -193,6 +195,8 @@ impl ControlPlaneCostModel { let dag = planner_types::post_asap::compile_executable_dag(root).ok()?; let mut value = 0.0; let mut states = 0; + let mut analytical = false; + let mut erp_record_ids = Vec::new(); for node in &dag.nodes { let family = match &node.payload { ExecutableOperatorPayload::SummaryAgg { family, .. } @@ -200,17 +204,38 @@ impl ControlPlaneCostModel { _ => continue, }; states += 1; - value += analytical_state_bytes(family)?; + let measurement = match family { + SummaryFamilyType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance) => self + .erp_costs + .as_ref() + .and_then(|erp| erp.candidate_memory_bytes(kind.algorithm(), kind.params())), + _ => None, + }; + if let Some((bytes, ids)) = measurement { + value += bytes; + erp_record_ids.extend(ids); + } else { + value += analytical_state_bytes(family)?; + analytical = true; + } } if states == 0 || !value.is_finite() { return None; } + erp_record_ids.sort(); + erp_record_ids.dedup(); Some(CandidateCostEstimate { value, unit: "bytes_per_state_partition", model: "backend_state_footprint_v1", - source: "analytical", - erp_record_ids: vec![], + source: if erp_record_ids.is_empty() { + "analytical" + } else if analytical { + "mixed" + } else { + "erp" + }, + erp_record_ids, }) } @@ -224,6 +249,8 @@ impl ControlPlaneCostModel { offline_frequency_comparison: None, exact_composition_costs: Vec::new(), erp: None, + erp_costs: None, + hll_confidence: None, } } @@ -235,11 +262,27 @@ impl ControlPlaneCostModel { self } + pub fn with_hll_confidence( + mut self, + contract: super::super::erp::HllConfidenceContract, + ) -> Self { + self.hll_confidence = Some(contract); + self + } + pub fn with_erp(mut self, erp: ErpPlanningInput) -> Self { + self.erp_costs = Some(erp.clone()); self.erp = Some(erp); self } + /// Resource evidence can remain usable when ERP's error observations + /// cannot establish the query's requested confidence guarantee. + pub fn with_erp_costs(mut self, erp: ErpPlanningInput) -> Self { + self.erp_costs = Some(erp); + self + } + pub fn erp_parameter_decision( &self, algorithm: SketchAlgorithm, @@ -384,6 +427,44 @@ impl ControlPlaneCostModel { costs.into_iter().map(|(algorithm, _)| algorithm).collect() } + fn rank_with_erp_costs( + &self, + intent: &AggIntent, + defaults: Vec, + ) -> Vec { + let Some(erp) = &self.erp_costs else { + return defaults; + }; + let (eps, delta) = + asap_aware_mapping::replacement::accuracy_budget(&intent_accuracy(intent)); + let mut has_measurement = false; + let mut costs = defaults + .iter() + .map(|algorithm| { + let params = self.size_params(algorithm.clone(), intent, eps, delta); + let measured = erp.candidate_memory_bytes(algorithm, ¶ms); + has_measurement |= measured.is_some(); + let cost = measured.map(|(bytes, _)| bytes).or_else(|| { + analytical_state_bytes(&SummaryFamilyType::Sketch( + planner_types::post_asap::SketchKind::new(algorithm.clone(), params), + GroupingStrategy::PerSubpopulationInstance, + )) + }); + (algorithm.clone(), cost) + }) + .collect::>(); + if !has_measurement { + return defaults; + } + costs.sort_by(|(_, a), (_, b)| match (a, b) { + (Some(a), Some(b)) => a.total_cmp(b), + (Some(_), None) => std::cmp::Ordering::Less, + (None, Some(_)) => std::cmp::Ordering::Greater, + (None, None) => std::cmp::Ordering::Equal, + }); + costs.into_iter().map(|(algorithm, _)| algorithm).collect() + } + /// Keep concrete physical identities in Planner's complete-candidate estimate, /// including distinct pane sizes using the same abstract window framework. pub fn with_window_implementation_costs( @@ -693,7 +774,7 @@ impl CostModel for ControlPlaneCostModel { // puts it first (`summary_candidates`), nothing to reorder. _ => candidates.to_vec(), }; - self.rank_with_offline_evidence(intent, defaults) + self.rank_with_erp_costs(intent, self.rank_with_offline_evidence(intent, defaults)) } fn size_params( @@ -703,6 +784,19 @@ impl CostModel for ControlPlaneCostModel { eps: f64, delta: f64, ) -> SketchParams { + if kind == SketchAlgorithm::Hll && matches!(intent, AggIntent::Cardinality { .. }) { + if let Some(contract) = &self.hll_confidence { + if let Some((eps, delta)) = self.combined_eps_delta(&intent_accuracy(intent)) { + // If no supported precision meets the target, keep the tightest + // candidate for explain; its guarantee still fails selection. + let precision = contract + .confidence(eps) + .and_then(|model| model.precision(delta)) + .unwrap_or(18); + return SketchParams::Hll { precision }; + } + } + } let (max_error, theoretical) = match intent { AggIntent::TopK { k, .. } => { let (eps, delta) = self.topk_eps_delta(&intent_accuracy(intent)); @@ -998,7 +1092,7 @@ fn hll_precision_for_eps(eps: f64) -> u8 { } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; use planner_types::pre_asap::{default_cardinality, default_quantile}; @@ -1006,6 +1100,153 @@ mod tests { AccuracyTarget::Epsilon(e) } + /// Logical summary selection has an explicit Planner estimate before + /// deployment pricing, including when a family is forced. + #[test] + fn summary_candidates_have_explicit_logical_costs() { + use asap_aware_mapping::{ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG}; + use std::rc::Rc; + + let accuracy = eps(0.1); + let root = Rc::new( + crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.9,m[1m])", + accuracy.clone(), + ) + .unwrap(), + ); + let target = TargetSubDAG::new(&root); + let model = ControlPlaneCostModel::new(accuracy.clone()); + let forced = ForcedFamilyCostModel::new(accuracy, SketchAlgorithm::DDSketch); + let candidates = SketchAlgorithmStrategy::new(&model).replacements(&target); + assert!(!candidates.is_empty()); + for candidate in &candidates { + let estimate = model.candidate_cost_estimate(candidate).unwrap(); + assert!(estimate.value.is_finite() && estimate.value > 0.0); + assert_eq!(estimate.source, "analytical"); + assert_eq!(estimate.unit, "bytes_per_state_partition"); + assert_eq!( + model.candidate_cost(candidate, &target), + Some(Cost(estimate.value)) + ); + assert_eq!( + forced.candidate_cost(candidate, &target), + Some(Cost(estimate.value)) + ); + } + } + + /// Analytical estimates scale with configured state size; unsupported + /// families remain unavailable instead of receiving a made-up zero. + #[test] + fn analytical_footprint_scales_with_parameters() { + use planner_types::post_asap::SketchKind; + let kll = |k| { + SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k }), + GroupingStrategy::PerSubpopulationInstance, + ) + }; + assert_eq!(analytical_state_bytes(&kll(200)), Some(6400.0)); + assert_eq!(analytical_state_bytes(&kll(400)), Some(12800.0)); + let unsupported = SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Theta, SketchParams::Theta { k: 1024 }), + GroupingStrategy::PerSubpopulationInstance, + ); + assert_eq!(analytical_state_bytes(&unsupported), None); + } + + pub(crate) fn erp_cost_fixture() -> ErpPlanningInput { + serde_json::from_value(serde_json::json!({ + "artifact": {"schema_version": 1, "producer_version": "synthetic-test-fixture", + "records": [{"id": "kll-200", "sketch": "kll-percall", "implementation": "lib", + "parameters": {"k": 200}, "distribution": {"test": "population-a"}, "trials": 20, + "error_metrics": {"max_rank_err": 0.9}, + "resources": {"memory_bytes": 7777.0, "update_cpu_seconds": 1e-7, + "merge_cpu_seconds": 1e-5, "query_cpu_seconds": 1e-6}}]}, + "distribution": {"test": "population-a"}, "implementation": "lib", + "error_metric": "max_rank_err", "min_trials": 10, + "expected_updates": 1000.0, "expected_queries": 100.0, "expected_merges": 1.0, + "retention_seconds": 60.0, "cpu_weight": 1.0, "byte_second_weight": 1e-9, + "mode": "hybrid" + })).unwrap() + } + + /// ERP memory overrides analytical memory for the exact configuration, + /// even when it is larger. Mismatched profiles fall back to analytical. + #[test] + fn erp_cost_precedes_analytical_without_certifying_accuracy() { + use asap_aware_mapping::{ReplacementStrategy, SketchAlgorithmStrategy}; + use std::rc::Rc; + let policy = erp_cost_fixture(); + let root = Rc::new( + crate::query_parser::parse_query_expr_canonical( + "quantile_over_time(0.9,m[1m])", + eps(0.1), + ) + .unwrap(), + ); + let model = ControlPlaneCostModel::new(eps(0.1)); + let candidates = + SketchAlgorithmStrategy::new(&model).replacements(&TargetSubDAG::new(&root)); + let candidate = candidates + .iter() + .find(|candidate| { + model + .candidate_cost_estimate(candidate) + .is_some_and(|cost| cost.value == 6400.0) + }) + .expect("KLL candidate"); + let estimate = |policy: ErpPlanningInput| { + ControlPlaneCostModel::new(eps(0.1)) + .with_erp_costs(policy) + .candidate_cost_estimate(candidate) + .unwrap() + }; + let matched = estimate(policy.clone()); + assert_eq!(matched.value, 7777.0); + assert_eq!(matched.source, "erp"); + assert_eq!(matched.erp_record_ids, ["kll-200"]); + let mut mismatch = policy.clone(); + mismatch.distribution = serde_json::json!({"test": "population-b"}); + assert_eq!(estimate(mismatch).source, "analytical"); + let mut mismatch = policy.clone(); + mismatch.implementation = Some("different-runtime".into()); + assert_eq!(estimate(mismatch).source, "analytical"); + let mut mismatch = policy.clone(); + mismatch.artifact.records[0].parameters = serde_json::json!({"k": 400}); + assert_eq!(estimate(mismatch).source, "analytical"); + let mut mismatch = policy.clone(); + mismatch.min_trials = 21; + assert_eq!(estimate(mismatch).source, "analytical"); + let mut mismatch = policy.clone(); + mismatch.artifact.records[0].resources.memory_bytes = f64::NAN; + assert_eq!(estimate(mismatch).source, "analytical"); + let model = ControlPlaneCostModel::new(eps(0.1)).with_erp_costs(policy); + assert_eq!( + model.rank_candidates( + &default_quantile(0.9), + &[SketchAlgorithm::DDSketch, SketchAlgorithm::Kll] + )[0], + SketchAlgorithm::Kll + ); + let (_, trace) = crate::planner_selection::select_workload_with_accuracy_model_and_trace( + vec![(0, root)], + eps(0.1), + &model, + &asap_aware_mapping::NoAccuracyEvidence, + &asap_aware_mapping::DefaultAccuracyModel, + ) + .unwrap(); + assert!(trace["groups"] + .as_array() + .unwrap() + .iter() + .flat_map(|group| group["candidates"].as_array().unwrap()) + .any(|candidate| candidate["cost_estimate"]["source"] == "erp" + && candidate["cost_estimate"]["unit"] == "bytes_per_state_partition")); + } + #[test] fn quantile_always_prefers_ddsketch_over_kll() { let model = ControlPlaneCostModel::new(AccuracyTarget::Epsilon(0.1)); diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index 963ebccac..8a7848717 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -4,7 +4,9 @@ //! Planner owns semantic legality. A manifest describes the exact physical //! demand to price; provider quotes and candidate evaluations are separate. +mod automatic; mod materialization_candidates; +pub use automatic::{AutomaticCostReport, ComponentResources}; mod status; pub use status::{CandidateEvaluationStatus, CandidateSearchScope}; @@ -90,6 +92,8 @@ pub struct CandidatePlanEvaluation { pub status: CandidateEvaluationStatus, pub plan_id: Option, pub total_cost: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automatic_cost: Option, pub unavailable_reason: Option, } @@ -500,6 +504,7 @@ fn candidate_description(candidate: &PhysicalCompilationRequest) -> CandidatePla status: CandidateEvaluationStatus::CompilationFailed, plan_id: None, total_cost: None, + automatic_cost: None, unavailable_reason: None, } } @@ -612,7 +617,7 @@ pub fn select_lowest_cost_candidate( select_candidates( candidates, env, - evidence, + Some(evidence), super::compiler::QueryFrontend::PromQl, ) } @@ -625,18 +630,20 @@ pub fn select_lowest_cost_metricsql_candidate( select_candidates( candidates, env, - evidence, + Some(evidence), super::compiler::QueryFrontend::MetricsQl, ) } -fn select_candidates( +pub fn select_candidates( candidates: Vec, env: PhysicalDeploymentContext, - evidence: &WorkloadCostEvidence, + evidence: Option<&WorkloadCostEvidence>, frontend: super::compiler::QueryFrontend, ) -> Result { - evidence.validate(&env)?; + if let Some(evidence) = evidence { + evidence.validate(&env)?; + } if candidates.is_empty() || candidates.len() > 64 { return Err(invalid( "candidate inventory must contain 1..=64 candidates", @@ -661,15 +668,23 @@ fn select_candidates( }); let planner_selection_trace = candidates[0].planner_selection_trace.clone(); let mut comparison_workload = None; + let mut comparison_inputs = None; let mut candidate_evaluations = Vec::new(); - let mut best_index = 0; - let mut best: Option<( - Cost, - CompiledPhysicalPlan, - WorkloadCostManifest, - BTreeMap, - )> = None; + let mut priced_candidates = Vec::new(); for candidate in candidates { + if evidence.is_none() { + let inputs = json!({"data":candidate.data_workload,"erp":candidate.erp}); + if comparison_inputs + .as_ref() + .is_some_and(|previous| previous != &inputs) + { + return Err(invalid( + "candidates describe different data/ERP cost inputs", + )); + } + comparison_inputs = Some(inputs); + } + let cost_request = candidate.clone(); let (plan, manifest, mut description) = match compile_candidate_for_pricing(candidate, env.clone(), frontend) { Ok(bound) => bound, @@ -686,19 +701,43 @@ fn select_candidates( return Err(invalid("candidates describe different workloads/horizons")); } comparison_workload = Some(scope); - match super::realization::RealizationProvider::price( - &super::realization::ExistingRealizations, - evidence, - &manifest, - ) { + let priced = if let Some(evidence) = evidence { + super::realization::RealizationProvider::price( + &super::realization::ExistingRealizations, + evidence, + &manifest, + ) + } else { + automatic::estimate(&cost_request, &env, &plan, &manifest) + .map(|report| { + let costs = report + .components + .iter() + .map(|(id, r)| (id.clone(), r.weighted_cost())) + .collect::>(); + let total = Cost(costs.values().sum()); + description.automatic_cost = Some(report); + (total, costs) + }) + .map_err(|error| { + ( + CandidateEvaluationStatus::EvidenceMissing, + error.to_string(), + ) + }) + }; + match priced { Ok((cost, components)) => { description.status = CandidateEvaluationStatus::Unselected; description.total_cost = Some(cost.0); candidate_evaluations.push(description); - if best.as_ref().is_none_or(|(previous, ..)| cost < *previous) { - best_index = candidate_evaluations.len() - 1; - best = Some((cost, plan, manifest, components)); - } + priced_candidates.push(Ok(( + cost, + plan, + manifest, + components, + candidate_evaluations.len() - 1, + ))); } Err((status, reason)) => { description.status = status; @@ -707,19 +746,37 @@ fn select_candidates( } } } - let (_, mut plan, selected_manifest, component_costs) = best.ok_or_else(|| { + let selected = asap_physical_operators::physical_planner::select_candidate( + priced_candidates, + |(cost, _, manifest, _, _)| { + Ok(Some( + asap_physical_operators::physical_planner::CandidateCost { + workload_scope: serde_json::to_string(&manifest.workload).map_err(|error| { + asap_physical_operators::Error::Invalid(error.to_string()) + })?, + horizon_seconds: manifest.horizon_seconds, + total_cost: cost.0, + }, + )) + }, + ) + .map_err(|error| { CompileError::Candidates( json!({"status": "all_infeasible", "logical_selection": planner_selection_trace, - "candidates": candidate_evaluations}), + "candidates": candidate_evaluations, "selection_error": error.to_string()}), ) })?; - // Exactly the winner retained by the existing strict-less-than selector. + let (_, mut plan, selected_manifest, component_costs, best_index) = selected.candidate; candidate_evaluations[best_index].status = CandidateEvaluationStatus::Selected; plan.cost_comparison = Some(CandidatePlanSelectionReport { planner_selection_trace, materialization_search_coverage, - data_snapshot_id: evidence.data_snapshot_id.clone(), - model_version: evidence.model_version.clone(), + data_snapshot_id: evidence + .map(|e| e.data_snapshot_id.clone()) + .unwrap_or_else(|| "request-scoped-data-workload".into()), + model_version: evidence + .map(|e| e.model_version.clone()) + .unwrap_or_else(|| automatic::MODEL_VERSION.into()), selected_plan_id: plan.envelope.plan_id, selected_manifest, component_costs, @@ -733,11 +790,7 @@ fn select_candidates( pub fn enumerate_exact_and_materialized_candidates( request: PhysicalCompilationRequest, ) -> Result, CompileError> { - let already_selected = super::maintained_population::supported(&request); let mut candidates = materialization_candidates(request)?; - if already_selected { - return Ok(candidates); - } let roots: Vec<_> = candidates .last() .expect("exact alternative") @@ -868,6 +921,257 @@ mod tests { use super::super::compiler::BackendLocalPlanningInput; use super::*; + /// A deployment without external execution rejects native candidates before pricing. + #[test] + fn local_only_deployment_rejects_external_candidate_before_pricing() { + let mut value = serde_json::to_value(fixture()).unwrap(); + value["implementation"]["require_backend_local_execution"] = json!(true); + let input: BackendLocalPlanningInput = serde_json::from_value(value).unwrap(); + let (request, environment) = input.clone().into_physical_compilation_request().unwrap(); + let candidates = enumerate_exact_and_materialized_candidates(request).unwrap(); + let native = candidates + .iter() + .find(|candidate| { + candidate.queries.iter().all(|query| { + matches!( + query.selected_plan_root.expr, + planner_types::post_asap::SummaryExpr::KeepPreAsap(_) + ) + }) + }) + .expect("native candidate remains inspectable") + .clone(); + let error = DeploymentPlanCompiler + .compile_promql(native, environment) + .unwrap_err(); + assert!( + error + .to_string() + .contains("external execution is unavailable"), + "{error}" + ); + let selected = input.compile_promql().unwrap(); + assert!(selected.query_plan.entries.values().all(|entry| entry + .nodes + .values() + .all(|node| !matches!(node, crate::query_plan::QueryPlanNode::ExactFallback { .. })))); + let report = selected.cost_comparison.unwrap(); + assert!(report + .candidate_evaluations + .iter() + .any(|candidate| candidate + .unavailable_reason + .as_ref() + .is_some_and(|reason| reason.contains("external execution is unavailable")) + && candidate.total_cost.is_none())); + } + + /// Selecting one population must not suppress candidates for other roots. + #[test] + fn existing_population_does_not_suppress_other_population_candidates() { + let mut input = fixture(); + let queries = input.query_workload.repeating_queries.as_mut().unwrap(); + let template = queries[0].clone(); + *queries = ["quantile by(job)(0.9,m)", "count by(job)(m)"] + .into_iter() + .map(|q| { + let mut entry = template.clone(); + entry.query = planner_types::workload::Query(q.into()); + entry + }) + .collect(); + let (mut request, _) = input.into_physical_compilation_request().unwrap(); + let strategy = asap_aware_mapping::maintained_population::MaintainedPopulationStrategy::new( + &request.canonical_roots, + ); + request.queries[0].selected_plan_root = + strategy.candidate(&request.canonical_roots[0]).unwrap(); + request.queries[1].selected_plan_root = + crate::planner_selection::keep_pre_asap(&request.canonical_roots[1]).unwrap(); + let candidates = enumerate_exact_and_materialized_candidates(request).unwrap(); + assert!(candidates + .iter() + .any(|candidate| candidate.queries.iter().all(|query| { + super::super::maintained_population::supported_node(&query.selected_plan_root) + }))); + } + + /// Instant counts select current membership, never accumulated observations. + #[test] + fn local_grouped_count_has_a_bindable_candidate() { + let mut input = fixture(); + input.workload_cost_evidence = None; + input.physical_inputs.require_backend_local_execution = true; + input.data_workload.data_ingestion_interval.value = + Some(planner_types::workload::DurationMs(60_000)); + input.physical_inputs.scrape_interval_ms = 60_000; + let queries = input.query_workload.repeating_queries.as_mut().unwrap(); + queries.truncate(1); + queries[0].query = planner_types::workload::Query("count by(job)(m)".into()); + let plan = input.compile_promql().unwrap(); + assert!(plan + .query_plan + .entries + .values() + .all(|entry| entry.nodes.values().any(|node| matches!( + node, + crate::query_plan::QueryPlanNode::Logical { + operator: crate::query_plan::residual::ResidualQueryOperator::CurrentSeries { + readout: asap_types::query_plan::current_series::SeriesReadout::Count, + .. + }, + .. + } + )))); + } + + /// Deployment computes and compares complete costs without external quotes. + #[test] + fn deployment_automatically_prices_workload() { + let mut input = fixture(); + input.workload_cost_evidence = None; + let plan = input.compile_promql().unwrap(); + let report = plan.cost_comparison.unwrap(); + let selected = report + .candidate_evaluations + .iter() + .find(|c| c.status == CandidateEvaluationStatus::Selected) + .unwrap(); + assert_eq!( + selected.total_cost.unwrap(), + report.component_costs.values().sum::() + ); + assert!(report + .candidate_evaluations + .iter() + .filter_map(|c| c.total_cost) + .all(|cost| cost >= selected.total_cost.unwrap())); + assert!(!report.component_costs.is_empty()); + assert_eq!( + report.component_costs.len(), + report.selected_manifest.components.len() + ); + assert!( + report + .candidate_evaluations + .iter() + .filter(|c| c.total_cost.is_some()) + .count() + > 1 + ); + } + + fn automatic_report( + request: &PhysicalCompilationRequest, + env: &PhysicalDeploymentContext, + plan: &CompiledPhysicalPlan, + ) -> AutomaticCostReport { + automatic::estimate( + request, + env, + plan, + &manifest(plan, &request.queries).unwrap(), + ) + .unwrap() + } + + /// Demand changes recurring work, while shared maintenance stays once per location. + #[test] + fn automatic_cost_scales_demand_data_and_shared_consumers() { + let (mut request, env) = fixture().into_physical_compilation_request().unwrap(); + let plan = DeploymentPlanCompiler + .compile_promql(request.clone(), env.clone()) + .unwrap(); + let before = automatic_report(&request, &env, &plan); + request.queries[0] + .summary_lifecycle_inputs + .evaluation_interval_ms /= 2; + let frequent = automatic_report(&request, &env, &plan); + for (id, cost) in &before.components { + let factor = if id.starts_with("query:") || id.starts_with("result:") { + 2.0 + } else { + 1.0 + }; + assert!( + (frequent.components[id].weighted_cost() - factor * cost.weighted_cost()).abs() + < 1e-10, + "{id}" + ); + } + request + .data_workload + .as_mut() + .unwrap() + .ingestion_rate + .value + .as_mut() + .unwrap() + .0 *= 2.0; + let larger = automatic_report(&request, &env, &plan); + assert!( + larger + .components + .iter() + .filter(|(id, _)| id.ends_with(":update")) + .map(|(_, c)| c.cpu_seconds) + .sum::() + > frequent + .components + .iter() + .filter(|(id, _)| id.ends_with(":update")) + .map(|(_, c)| c.cpu_seconds) + .sum::() + ); + let mut shared = plan.clone(); + let mut query = request.queries[0].clone(); + query.query_id = "second-consumer".into(); + let mut entry = shared.query_plan.entries.values().next().unwrap().clone(); + entry.query_id = query.query_id.clone(); + shared + .query_plan + .entries + .insert(query.query_id.clone(), entry); + request.queries.push(query); + let twice = automatic_report(&request, &env, &shared); + for (id, cost) in &larger.components { + assert_eq!(&twice.components[id], cost); + } + assert_eq!( + twice + .components + .keys() + .filter(|id| id.starts_with("source:") || id.starts_with("state:")) + .count(), + larger + .components + .keys() + .filter(|id| id.starts_with("source:") || id.starts_with("state:")) + .count() + ); + } + + /// Expired facts and arithmetic overflow cannot silently become a zero quote. + #[test] + fn automatic_cost_rejects_stale_and_overflowing_data() { + let (mut request, env) = fixture().into_physical_compilation_request().unwrap(); + let plan = DeploymentPlanCompiler + .compile_promql(request.clone(), env.clone()) + .unwrap(); + let manifest = manifest(&plan, &request.queries).unwrap(); + let rate = &mut request.data_workload.as_mut().unwrap().ingestion_rate; + rate.observed_at_ms = Some(1); + rate.valid_for_ms = Some(1); + assert!(automatic::estimate(&request, &env, &plan, &manifest) + .unwrap_err() + .to_string() + .contains("fresh ingestion rate")); + let rate = &mut request.data_workload.as_mut().unwrap().ingestion_rate; + rate.valid_for_ms = None; + rate.value.as_mut().unwrap().0 = f64::MAX; + assert!(automatic::estimate(&request, &env, &plan, &manifest).is_err()); + } + fn fixture() -> BackendLocalPlanningInput { let mut snapshot: BackendLocalPlanningInput = serde_json::from_str(include_str!( "../../../docs/examples/asapquery-planning-snapshot.json" @@ -1383,9 +1687,14 @@ mod tests { } #[test] - fn snapshot_requires_quotes_and_roundtrips_selection() { + fn snapshot_supports_automatic_cost_and_roundtrips_quote_override() { let mut snapshot = fixture(); - assert!(snapshot.clone().compile_promql().is_err()); + assert!(snapshot + .clone() + .compile_promql() + .unwrap() + .cost_comparison + .is_some()); let (_, _, evidence) = quoted(); snapshot.workload_cost_evidence = Some(evidence); let snapshot: BackendLocalPlanningInput = diff --git a/control_plane/src/physical/workload_cost/automatic.rs b/control_plane/src/physical/workload_cost/automatic.rs new file mode 100644 index 000000000..127b90d18 --- /dev/null +++ b/control_plane/src/physical/workload_cost/automatic.rs @@ -0,0 +1,667 @@ +//! Versioned reference resource model. These coefficients are analytical +//! assumptions, not measurements or currency prices. ERP replaces only the +//! resource dimensions it measures; plan demand always comes from the backend. +use super::*; +use crate::query_plan::{residual::ResidualQueryOperator as Op, QueryPlanNode as Node}; +use asap_aware_mapping::erp::ErpResourceProfile; +use asap_types::{ + AggregationType as A, PrecomputeMaterialization, WindowMaterializationLayout as Layout, +}; +use planner_types::post_asap::SketchAlgorithm; + +pub const MODEL_VERSION: &str = "backend-workload-resources-v1"; +const CPU_PER_ITEM: f64 = 1e-7; +const CPU_PER_BYTE: f64 = 1e-9; +const SAMPLE_BYTES: f64 = 24.0; +const SERIES_BYTES: f64 = 256.0; +const MEMORY_WEIGHT: f64 = 1e-9; +const NETWORK_WEIGHT: f64 = 1e-8; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ComponentResources { + pub cpu_seconds: f64, + pub memory_byte_seconds: f64, + pub network_bytes: f64, + pub source: String, + pub erp_record_ids: Vec, + pub calculation: Value, +} +impl ComponentResources { + pub(super) fn weighted_cost(&self) -> f64 { + self.cpu_seconds + + self.memory_byte_seconds * MEMORY_WEIGHT + + self.network_bytes * NETWORK_WEIGHT + } +} +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct AutomaticCostReport { + pub model_version: String, + pub weights: Value, + pub inputs: Value, + pub assumptions: Vec, + pub components: BTreeMap, +} +#[derive(Clone)] +struct State { + profile: ErpResourceProfile, + ids: Vec, + partitions: f64, + retained: f64, + allocations_per_second: f64, + overlap: f64, + rollup_merges: f64, +} +fn resources( + cpu: f64, + memory: f64, + network: f64, + ids: Vec, + calculation: Value, +) -> ComponentResources { + ComponentResources { + cpu_seconds: cpu, + memory_byte_seconds: memory, + network_bytes: network, + source: if ids.is_empty() { + "analytical" + } else { + "erp+analytical" + } + .into(), + erp_record_ids: ids, + calculation, + } +} +fn state( + m: &PrecomputeMaterialization, + request: &PhysicalCompilationRequest, + cardinality: u64, +) -> Result { + let bytes = + super::super::compiler::estimated_state_bytes(&m.aggregation_type, &m.parameters) as f64; + let mut profile = ErpResourceProfile { + memory_bytes: bytes, + update_cpu_seconds: CPU_PER_ITEM * (bytes / 16.0).max(2.0).log2(), + merge_cpu_seconds: CPU_PER_BYTE * bytes, + query_cpu_seconds: CPU_PER_BYTE * bytes + CPU_PER_ITEM, + }; + let mut ids = Vec::new(); + let algorithm = match m.aggregation_type { + A::DatasketchesKLL => Some(SketchAlgorithm::Kll), + A::HLL => Some(SketchAlgorithm::Hll), + A::UnivMon => Some(SketchAlgorithm::UnivMon), + _ => None, + }; + if let (Some(policy), Some(algorithm)) = (&request.erp, algorithm) { + let mut policy = policy.clone(); + // Use the same deployed implementation and population checks as the + // logical cost path. Unknown implementations fall back to this model. + policy.artifact.records.retain(|row| { + (row.sketch == "kll-percall" && row.implementation == "lib") + || (row.sketch == "hll" && row.implementation == "asap-sketchlib-hll-regular-v1") + || (row.sketch == "univmon" + && row.implementation == "asap-sketchlib-univmon-standard-v1") + }); + let population_matches = policy.observed_populations.is_none() + || (!request.canonical_roots.is_empty() + && request.canonical_roots.iter().all(|root| { + super::super::compiler::observed_population_matches_root(&policy, root) + })); + if population_matches { + if let Some(params) = super::super::erp::parse_params(&algorithm, &json!(m.parameters)) + { + if let Some((measured, records)) = policy.candidate_resources(&algorithm, ¶ms) { + profile = measured; + ids = records; + } + } + } + } + let (period, overlap, rollup_merges) = match &m.window_layout { + Layout::Pane { pane_secs } => (*pane_secs as f64, 1.0, 0.0), + Layout::FullWindow => ( + m.slide_interval as f64, + (m.window_size as f64 / m.slide_interval as f64).ceil(), + 0.0, + ), + Layout::HierarchicalRollup { + base_pane_secs, + levels_secs, + } => { + // Every closed child state is folded into its parent once. + let periods = std::iter::once(base_pane_secs) + .chain(levels_secs.iter().take(levels_secs.len().saturating_sub(1))); + ( + *base_pane_secs as f64, + 1.0, + periods.map(|p| 1.0 / *p as f64).sum(), + ) + } + }; + let retained = m + .num_aggregates_to_retain + .ok_or_else(|| invalid("state retention count is unknown"))? as f64; + if period <= 0.0 || retained <= 0.0 { + return Err(invalid("invalid state layout")); + } + Ok(State { + profile, + ids, + partitions: super::super::compiler::retained_partition_count(m, Some(cardinality)) as f64, + retained, + allocations_per_second: 1.0 / period + + match &m.window_layout { + Layout::HierarchicalRollup { levels_secs, .. } => { + levels_secs.iter().map(|p| 1.0 / *p as f64).sum::() + } + _ => 0.0, + }, + overlap, + rollup_merges, + }) +} + +pub(super) fn estimate( + request: &PhysicalCompilationRequest, + env: &PhysicalDeploymentContext, + plan: &CompiledPhysicalPlan, + manifest: &WorkloadCostManifest, +) -> Result { + let data = request + .data_workload + .as_ref() + .ok_or_else(|| invalid("automatic costing needs data_workload"))?; + let now = env.observed_at_unix_ms; + let horizon = manifest.horizon_seconds; + let rate = data + .ingestion_rate + .value_at(now) + .map(|r| r.0) + .ok_or_else(|| invalid("automatic costing needs a fresh ingestion rate"))?; + let cadence = data + .data_ingestion_interval + .value_at(now) + .map(|d| d.0 as f64 / 1000.0) + .ok_or_else(|| invalid("automatic costing needs fresh source cadence"))?; + if !rate.is_finite() || rate < 0.0 || !cadence.is_finite() || cadence <= 0.0 { + return Err(invalid("invalid source rate/cadence")); + } + let mut assumptions = vec![ + "Reference analytical coefficients are estimates, not calibrated benchmarks or currency prices.".into(), + "Absent per-source/selectivity facts, each source and collector uses the whole workload rate and cardinality (conservative replication).".into(), + "Grouped states use input cardinality as an upper bound on group count; filtered inputs receive no selectivity discount.".into(), + "Transport uses full state size even for deltas; memory footprint approximates encoded size; result rows include a 256-byte label allowance.".into(), + ]; + let cardinality = match data.input_cardinality.value_at(now) { + Some(value) => *value, + None => { + if rate == 0.0 { + return Err(invalid( + "zero arrival rate cannot establish active cardinality", + )); + } + assumptions.push("Unknown cardinality estimated as ceil(ingestion_rate * source_cadence), assuming one sample per active series per scrape.".into()); + let n = (rate * cadence).ceil(); + if !n.is_finite() || n >= u64::MAX as f64 { + return Err(invalid("cardinality estimate overflow")); + } + n as u64 + } + }; + if cardinality == 0 && rate > 0.0 { + return Err(invalid("positive ingestion rate with zero cardinality")); + } + let cardinality = cardinality as f64; + let mut states = BTreeMap::new(); + for schema in &plan.precompute_plan.schemas { + let m = plan + .precompute_plan + .materializations + .iter() + .find(|m| m.policy_fingerprint() == schema.materialization.fingerprint()) + .ok_or_else(|| invalid("missing materialization"))?; + states.insert( + schema.materialization.0, + state(m, request, cardinality as u64)?, + ); + } + let mut components = BTreeMap::new(); + let mut insert = |id: String, value: ComponentResources| -> Result<(), CompileError> { + if !manifest.components.contains_key(&id) { + return Err(invalid(format!("unmanifested cost {id}"))); + } + if [ + value.cpu_seconds, + value.memory_byte_seconds, + value.network_bytes, + value.weighted_cost(), + ] + .iter() + .any(|v| !v.is_finite() || *v < 0.0) + { + return Err(invalid(format!("non-finite/negative resource cost {id}"))); + } + if components.insert(id.clone(), value).is_some() { + return Err(invalid(format!("duplicate cost {id}"))); + } + Ok(()) + }; + let updates = rate * horizon; + let max_lookback = plan + .query_plan + .entries + .values() + .map(|e| e.instant.lookback_ms as f64 / 1000.0) + .fold(cadence, f64::max); + let needs_volume = matches!(data.arrival, planner_types::workload::DataArrival::AtRest) + || plan + .query_plan + .entries + .values() + .any(|e| e.instant.full_history); + let source_samples = + if needs_volume { + *data.ingestion_volume.value_at(now).ok_or_else(|| { + invalid("at-rest/full-history costing needs fresh ingestion_volume") + })? as f64 + + updates + } else { + rate * max_lookback + }; + for (id, demand) in &manifest.components { + if id.starts_with("source:") { + let location = demand.implementation["location"].as_str().unwrap_or(""); + let remote_backend = location == "backend" && !plan.collector_plans.is_empty(); + let items = if remote_backend { 0.0 } else { updates }; + let raw_retained = if location == "exact_backend" { + source_samples * SAMPLE_BYTES + cardinality * SERIES_BYTES + } else { + 0.0 + }; + insert( + id.clone(), + resources( + items * CPU_PER_ITEM, + raw_retained * horizon, + items * SAMPLE_BYTES, + vec![], + json!({"input_samples": items, "retained_raw_bytes": raw_retained, "cpu_seconds_per_sample": CPU_PER_ITEM, + "backend_summary_decode_charged_in_transport": remote_backend}), + ), + )?; + } else if id.starts_with("state:") { + let binding = &demand.implementation["binding"]; + let materialization: asap_types::PolicyFingerprint = + serde_json::from_value(binding["schema"]["materialization"].clone()) + .map_err(|e| invalid(e.to_string()))?; + let s = states + .get(&materialization) + .ok_or_else(|| invalid("unknown state"))?; + let operation = demand.implementation["operation"].as_str().unwrap_or(""); + let builds = s.partitions * (s.retained + (horizon * s.allocations_per_second).ceil()); + let remote_backend = + binding["location"] == "backend" && !plan.collector_plans.is_empty(); + let update_count = if remote_backend { + 0.0 + } else { + updates * s.overlap + }; + let merges = s.partitions * horizon * s.rollup_merges; + let (cpu, memory) = match operation { + "build" => (builds * s.profile.memory_bytes * CPU_PER_BYTE, 0.0), + "update" => ( + update_count * s.profile.update_cpu_seconds + + merges * s.profile.merge_cpu_seconds, + 0.0, + ), + "residency" => ( + 0.0, + s.partitions * s.retained * s.profile.memory_bytes * horizon, + ), + "retire" => (builds * CPU_PER_ITEM, 0.0), + _ => return Err(invalid("unsupported state phase")), + }; + insert( + id.clone(), + resources( + cpu, + memory, + 0.0, + s.ids.clone(), + json!({"operation":operation, + "partitions":s.partitions, "retained_states_per_partition":s.retained, "builds_and_retirements":builds, + "updates":update_count,"rollup_merges":merges,"unit_resources":s.profile, + "allocation_cpu_seconds_per_byte":CPU_PER_BYTE,"retirement_cpu_seconds_per_state":CPU_PER_ITEM}), + ), + )?; + } else if id.starts_with("current-series:") { + let phase = demand.implementation["phase"].as_str().unwrap_or(""); + let bytes = cardinality * SERIES_BYTES; + let retention_ms = demand.implementation["population"]["history_retention_ms"] + .as_u64() + .unwrap_or(0); + let versions = if retention_ms == 0 { + 0.0 + } else { + (retention_ms as f64 / (cadence * 1000.)).ceil() + 1.0 + }; + let retained_bytes = bytes + versions * (bytes + 1024.0); + let snapshots = if versions == 0.0 { + 0.0 + } else { + horizon / cadence + }; + let (cpu, memory) = match phase { + "build" => (bytes * CPU_PER_BYTE, 0.0), + "update" => ( + updates * CPU_PER_ITEM * cardinality.max(2.0).log2() + + snapshots * bytes * CPU_PER_BYTE, + 0.0, + ), + "residency" => (0.0, retained_bytes * horizon), + "retire" => ((cardinality + snapshots) * CPU_PER_ITEM, 0.0), + _ => return Err(invalid("unsupported current-series phase")), + }; + insert( + id.clone(), + resources( + cpu, + memory, + 0.0, + vec![], + json!({"phase":phase,"series":cardinality,"updates":updates,"bytes_per_series":SERIES_BYTES,"historical_versions":versions,"retained_bytes":retained_bytes,"snapshots":snapshots}), + ), + )?; + } + } + for rule in &plan.transmission_plan.rules { + if rule.emit_every_ms == 0 { + return Err(invalid("unknown transmission cadence")); + } + let s = states + .get(&rule.materialization.0) + .ok_or_else(|| invalid("unknown transport state"))?; + let checkpoints = rule + .full_checkpoint_every_ms + .map(|ms| { + if ms == 0 { + f64::INFINITY + } else { + horizon * 1000.0 / ms as f64 + } + }) + .unwrap_or(0.0); + let frames = (horizon * 1000.0 / rule.emit_every_ms as f64 + checkpoints).ceil() + * s.partitions + * s.retained; + let bytes = frames * s.profile.memory_bytes; + insert( + format!("transport:{}:{}", rule.producer_id, rule.materialization.0), + resources( + bytes * CPU_PER_BYTE * 2.0 + frames * s.profile.merge_cpu_seconds, + 0.0, + bytes, + s.ids.clone(), + json!({"frames":frames,"bytes_per_frame":s.profile.memory_bytes,"merge_cpu_seconds_per_frame":s.profile.merge_cpu_seconds, + "encode_decode_cpu_seconds_per_byte":2.0*CPU_PER_BYTE}), + ), + )?; + } + for entry in plan.query_plan.entries.values() { + let mut rows = BTreeMap::new(); + let mut profiles: BTreeMap<_, Vec<&State>> = BTreeMap::new(); + for node_id in entry + .topological_order() + .map_err(|e| invalid(e.to_string()))? + { + let node = &entry.nodes[&node_id]; + let id = format!("query:{}:{}", entry.query_id, node_id.0); + let evaluations = manifest.components[&id].occurrences_per_horizon; + let input_rows: f64 = node.inputs().iter().map(|id| rows[id]).sum(); + let mut output_rows = input_rows.max(1.0); + let mut used: Vec<&State> = node + .inputs() + .iter() + .flat_map(|id| profiles.get(id).into_iter().flatten().copied()) + .collect(); + let mut detail = json!({"input_rows":input_rows}); + let mut network = 0.0; + let cpu = match node { + Node::ReadMaterialization { binding } => { + let s = states + .get(&binding.materialization.0) + .ok_or_else(|| invalid("unknown read state"))?; + let panes = if binding.full_window_slide_ms.is_some() { + 1.0 + } else { + (binding + .readout_lookback_ms + .unwrap_or(entry.instant.lookback_ms) as f64 + / binding.window_ms as f64) + .ceil() + .max(1.0) + }; + output_rows = s.partitions; + used = vec![s]; + detail = json!({"partitions":s.partitions,"panes_per_read":panes,"unit_resources":s.profile}); + s.partitions + * (panes * s.profile.memory_bytes * CPU_PER_BYTE + + (panes - 1.0) * s.profile.merge_cpu_seconds) + } + Node::SummaryEstimate { .. } => used + .iter() + .map(|s| s.partitions * s.profile.query_cpu_seconds) + .sum(), + Node::SummaryMerge { .. } => used + .iter() + .map(|s| s.partitions * s.profile.merge_cpu_seconds) + .sum(), + Node::Scalar { .. } => { + output_rows = 1.0; + CPU_PER_ITEM + } + Node::ExactFallback { .. } + | Node::Logical { + operator: Op::ExactSubquery { .. } | Op::CandidateExactSubquery { .. }, + .. + } => { + let query = match node { + Node::Logical { + operator: + Op::ExactSubquery { query } | Op::CandidateExactSubquery { query, .. }, + .. + } => query.as_str(), + _ => entry.canonical_query.as_str(), + }; + let parsed = crate::query_parser::parse_query_expr_canonical( + query, + crate::types::AccuracyTarget::Exact, + ) + .map_err(|e| invalid(e.to_string()))?; + let sources = exact_source_metrics(&parsed)?.len() as f64; + let samples = if needs_volume { + *data.ingestion_volume.value_at(now).ok_or_else(|| { + invalid("full-history exact cost needs fresh ingestion_volume") + })? as f64 + } else { + (rate * (entry.instant.lookback_ms as f64 / 1000.0).max(cadence)) + .max(cardinality) + } * sources; + let operators = tree_size( + &serde_json::to_value(parsed).map_err(|e| invalid(e.to_string()))?, + ) as f64; + output_rows = cardinality.max(1.0); + network = output_rows * SERIES_BYTES + query.len() as f64; + detail = json!({"scanned_samples":samples,"syntax_objects":operators,"cpu_seconds_per_item":CPU_PER_ITEM, + "formula":"(samples + 1) * log2(max(samples, 2)) * syntax_objects * cpu_seconds_per_item"}); + (samples + 1.0) * samples.max(2.0).log2() * operators * CPU_PER_ITEM + } + Node::Logical { + operator: Op::CurrentSeries { .. }, + .. + } => { + output_rows = cardinality; + cardinality * CPU_PER_ITEM + } + Node::RelationalJoin { + inputs, + join_kind, + pred, + left_schema, + right_schema, + .. + } if matches!( + join_kind, + planner_types::pre_asap::JoinKind::Semi + | planner_types::pre_asap::JoinKind::Anti + ) => + { + output_rows = rows[&inputs[0]]; + let indexed = *join_kind == planner_types::pre_asap::JoinKind::Semi + && serde_json::from_value(pred.clone()).is_ok_and(|predicate| { + asap_physical_operators::dag::planner::equijoin_keys( + &predicate, + left_schema, + right_schema, + ) + .is_ok() + }); + detail = json!({"input_rows": input_rows, "output_rows_upper_bound": output_rows, "indexed_equality_semijoin": indexed}); + if indexed { + input_rows.max(1.0) * input_rows.max(2.0).log2() * CPU_PER_ITEM + } else { + rows[&inputs[0]] * rows[&inputs[1]] * CPU_PER_ITEM + } + } + Node::RelationalJoin { .. } => { + output_rows = input_rows.powi(2); + output_rows * CPU_PER_ITEM + } + Node::ExternalExact { .. } + | Node::Logical { + operator: Op::Scan { .. } | Op::Subquery { .. }, + .. + } => { + return Err(invalid("no automatic model for external exact/generic scan/nested subquery operator")); + } + Node::ReduceSum { grouping, .. } => { + if matches!(grouping, crate::query_plan::PhysicalGrouping::Reduce(keys) if keys.is_empty()) + { + output_rows = 1.0; + } + input_rows * CPU_PER_ITEM + } + Node::ExactReadout { .. } + | Node::Binary { .. } + | Node::Relational { .. } + | Node::Logical { .. } => { + input_rows.max(1.0) * input_rows.max(2.0).log2() * CPU_PER_ITEM + } + }; + let ids: BTreeSet<_> = if matches!( + node, + Node::ReadMaterialization { .. } + | Node::SummaryEstimate { .. } + | Node::SummaryMerge { .. } + ) { + used.iter().flat_map(|s| s.ids.iter().cloned()).collect() + } else { + BTreeSet::new() + }; + detail["evaluations"] = json!(evaluations); + detail["output_rows_bound"] = json!(output_rows); + insert( + id, + resources( + cpu * evaluations, + 0.0, + network * evaluations, + ids.into_iter().collect(), + detail, + ), + )?; + rows.insert(node_id, output_rows); + profiles.insert(node_id, used); + } + let id = format!("result:{}", entry.query_id); + let evaluations = manifest.components[&id].occurrences_per_horizon; + let bytes = rows[&entry.root] * SERIES_BYTES * evaluations; + insert( + id, + resources( + bytes * CPU_PER_BYTE, + 0.0, + bytes, + vec![], + json!({"rows_per_evaluation":rows[&entry.root],"evaluations":evaluations,"bytes_per_row":SERIES_BYTES}), + ), + )?; + } + if components.len() != manifest.components.len() { + return Err(invalid( + "automatic cost model did not cover every manifest component", + )); + } + if !components + .values() + .map(ComponentResources::weighted_cost) + .sum::() + .is_finite() + { + return Err(invalid("total workload cost overflow")); + } + Ok(AutomaticCostReport { + model_version: MODEL_VERSION.into(), + weights: json!({"unit":"weighted_resource_seconds","cpu_seconds":1.0,"memory_byte_seconds":MEMORY_WEIGHT,"network_bytes":NETWORK_WEIGHT}), + inputs: json!({"data_workload":data,"horizon_seconds":horizon,"ingestion_rate":rate,"source_cadence_seconds":cadence, + "input_cardinality":cardinality,"capability_snapshot_id":env.capability_snapshot_id,"observed_at_unix_ms":now}), + assumptions, + components, + }) +} +fn tree_size(value: &Value) -> usize { + match value { + Value::Object(fields) => 1 + fields.values().map(tree_size).sum::(), + Value::Array(items) => items.iter().map(tree_size).sum(), + _ => 0, + } +} + +#[cfg(test)] +mod tests { + use super::*; + /// A concrete implementation uses all ERP resources even when dearer; + /// a profile for a different implementation cannot replace its estimate. + #[test] + fn physical_state_resources_prefer_applicable_erp() { + let input: super::super::super::compiler::BackendLocalPlanningInput = serde_json::from_str( + include_str!("../../../../docs/examples/asapquery-planning-snapshot.json"), + ) + .unwrap(); + let (mut request, env) = input.into_physical_compilation_request().unwrap(); + let plan = super::super::super::compiler::DeploymentPlanCompiler + .compile_promql(request.clone(), env) + .unwrap(); + let mut m = plan.precompute_plan.materializations[0].clone(); + m.aggregation_type = A::DatasketchesKLL; + m.parameters = std::collections::HashMap::from([("k".into(), json!(200))]); + let baseline = state(&m, &request, 100).unwrap(); + let mut erp = crate::physical::post_asap::cost_model::tests::erp_cost_fixture(); + erp.artifact.records[0].resources = ErpResourceProfile { + memory_bytes: 1e8, + update_cpu_seconds: 0.1, + merge_cpu_seconds: 0.2, + query_cpu_seconds: 0.3, + }; + let expected = erp.artifact.records[0].resources.clone(); + request.erp = Some(erp); + let measured = state(&m, &request, 100).unwrap(); + assert!(!measured.ids.is_empty()); + assert_eq!(measured.profile, expected); + assert!(measured.profile.memory_bytes > baseline.profile.memory_bytes); + request.erp.as_mut().unwrap().artifact.records[0].implementation = "other-runtime".into(); + let fallback = state(&m, &request, 100).unwrap(); + assert!(fallback.ids.is_empty()); + assert_eq!(fallback.profile, baseline.profile); + } +} diff --git a/control_plane/src/planner_selection.rs b/control_plane/src/planner_selection.rs index 37e17fd7f..a97342bb1 100644 --- a/control_plane/src/planner_selection.rs +++ b/control_plane/src/planner_selection.rs @@ -1,8 +1,8 @@ -//! Deployment-owned selection at the latest ASAPPlanner boundary. +//! Supplies deployment capabilities and cost/accuracy evidence to ASAPPlanner. //! -//! ASAPPlanner enumerates a ranked candidate space and deliberately does not -//! commit to one deployment plan. The backend owns that decision because it -//! also owns placement, runtime capabilities, and the physical wire contract. +//! Planner constructs, evaluates, and selects computation candidates. This +//! adapter registers the supported strategies and retains selection evidence; +//! DeploymentPlanCompiler binds the resulting computation and lifecycle. use std::rc::Rc; diff --git a/control_plane/tests/discovery_snapshot.rs b/control_plane/tests/discovery_snapshot.rs index c6e4a69b0..ecedfd3fa 100644 --- a/control_plane/tests/discovery_snapshot.rs +++ b/control_plane/tests/discovery_snapshot.rs @@ -55,7 +55,7 @@ fn discovered_snapshot_plans_with_observed_cadence_and_promql_history() { .into_physical_compilation_request() .unwrap(); assert_eq!(request.scrape_interval_ms, Some(60_000)); - assert_eq!(request.queries[0].query_lookback_seconds, 3660); + assert_eq!(request.queries[0].query_lookback_ms, 3_660_000); let roundtrip: BackendLocalPlanningInput = serde_json::from_value(serde_json::to_value(&snapshot).unwrap()).unwrap(); assert_eq!(snapshot, roundtrip); diff --git a/data_plane/tests/support/distinct_planning_process.rs b/data_plane/tests/support/distinct_planning_process.rs index 72bb3dd74..242844ec5 100644 --- a/data_plane/tests/support/distinct_planning_process.rs +++ b/data_plane/tests/support/distinct_planning_process.rs @@ -1,4 +1,5 @@ use super::*; +use control_plane::physical::compiler::BackendLocalPlanningInput; /// Modeled HLL error without a confidence certificate cannot replace exact execution. #[tokio::test] @@ -14,3 +15,181 @@ async fn uncertified_distinct_uses_exact_process() { fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); assert_uncertified_exact_process(fixture, &[QUERY]).await; } + +/// The production compiler, ingest engine and query DAG preserve distinct populations. +#[tokio::test] +async fn bounded_classic_hll_executes_with_source_labels() { + const QUERY: &str = "distinct_over_time(distinct_values{job=\"api\"}[5s])"; + let fallback_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let fallback_url = format!("http://{}", fallback_listener.local_addr().unwrap()); + let fallback_task = tokio::spawn(async move { + axum::serve( + fallback_listener, + Router::new().route("/-/healthy", get(|| async { "healthy" })), + ) + .await + .unwrap(); + }); + let mut fixture: Value = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let mut entry = fixture["query_workload"]["repeating_queries"][3].clone(); + entry["query"] = QUERY.into(); + entry["requirements"]["accuracy"] = serde_json::json!({"explicit": {"Epsilon": 0.05}}); + fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); + // The generated finite value domain enforces this bound over complete + // readout populations, including all panes. Series count is not the bound. + fixture["implementation"]["data_snapshot_id"] = "process-fixture".into(); + fixture["implementation"]["accuracy_evidence"] = serde_json::json!({QUERY: { + "query_string": QUERY, "data_snapshot_id": "process-fixture", + "data_workload": fixture["data_workload"], "source": "finite-generated-value-domain", + "observed_at_unix_ms": fixture["environment"]["observed_at_unix_ms"], + "valid_for_ms": 60000, + "hll": {"model":"asap-classic64-uniform-hash-linear-counting-v1", + "max_distinct_per_readout":128} + }}); + let plan = quote_snapshot_for_test( + serde_json::from_value::(fixture.clone()).unwrap(), + ) + .compile_promql() + .unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 1); + assert_eq!( + plan.precompute_plan.materializations[0].aggregation_type, + asap_types::AggregationType::HLL + ); + eprintln!( + "DISTINCT_PLANNED {}", + serde_json::json!({"materializations": plan.precompute_plan.materializations, "query_plan": plan.query_plan, "lifecycle_estimates": plan.lifecycle_estimates}) + ); + let output = tempfile::tempdir().unwrap(); + let path = output.path().join("planning.json"); + let priced = quote_snapshot_for_test(serde_json::from_value(fixture.clone()).unwrap()); + std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap(); + let port = unused_port(); + let mut vm_port = unused_port(); + while vm_port == port { + vm_port = unused_port(); + } + let mut child = ChildGuard( + Command::new(env!("CARGO_BIN_EXE_data_plane")) + .args([ + "--forward-unsupported-queries", + "--prometheus-server", + &fallback_url, + "--profile", + "asapquery", + "--planning-snapshot", + ]) + .arg(&path) + .args(["--http-port", &port.to_string(), "--output-dir"]) + .arg(output.path()) + .args([ + "--victoriametrics-http-port", + &vm_port.to_string(), + "--victoriametrics-url", + &fallback_url, + ]) + .args([ + "--precompute-allowed-lateness-ms", + "0", + "--precompute-flush-interval-ms", + "25", + ]) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(), + ); + let client = reqwest::Client::new(); + let backend = format!("http://127.0.0.1:{port}"); + wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await; + // Source syntax uses the shared parser fork; serving semantics and exact + // routing belong to the MetricsQL adapter and its installed query entries. + let snapshot = serde_json::from_value::(fixture).unwrap(); + let mut snapshot = snapshot; + snapshot.environment.plan_version = 2; + let compiled = quote_snapshot_for_frontend_test(snapshot, true) + .compile_metricsql() + .unwrap(); + let identity = serde_json::json!({"plan_id": compiled.envelope.plan_id, "plan_version": compiled.envelope.plan_version}); + let install = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { + summary_catalog: compiled.summary_catalog, + collector_plans: compiled.collector_plans, + precompute_plan: compiled.precompute_plan, + transmission_plan: compiled.transmission_plan, + query_plan: compiled.query_plan, + storage_routing: None, + adaptation_evidence: vec![], + }; + eprintln!( + "DISTINCT_INSTALLED {}", + serde_json::to_string(&install).unwrap() + ); + let response = client + .post(format!("{backend}/api/v1/physical-plan")) + .json(&install) + .send() + .await + .unwrap(); + assert!( + response.status().is_success(), + "{}", + response.text().await.unwrap() + ); + let response = client + .post(format!("{backend}/api/v1/physical-plan/activate")) + .json(&identity) + .send() + .await + .unwrap(); + assert!( + response.status().is_success(), + "{}", + response.text().await.unwrap() + ); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as i64; + let base = now - now.rem_euclid(5000) - 20000; + let mut series = Vec::new(); + for (instance, distinct, job) in [("a", 5, "api"), ("b", 13, "api"), ("excluded", 23, "other")] + { + let mut samples: Vec<_> = (0..100) + .map(|i| (base + 1 + i, (i % distinct) as f64)) + .collect(); + samples.push((base + 15001, 1000.0)); + series.push(series_with_labels( + "distinct_values", + &[("instance", instance), ("job", job)], + &samples, + )); + } + assert_eq!( + remote_write(&client, &backend, &WriteRequest { timeseries: series }).await, + 204 + ); + drain_precompute(&client, &backend).await; + let result = wait_for_warm_instant( + &client, + &format!("http://127.0.0.1:{vm_port}"), + QUERY, + (base + 5000) as f64 / 1000.0, + &output.path().join("query_engine.log"), + ) + .await; + let rows = result["data"]["result"].as_array().unwrap(); + assert_eq!(rows.len(), 2, "{result}"); + for (instance, exact) in [("a", 5.0), ("b", 13.0)] { + let row = rows + .iter() + .find(|row| row["metric"]["instance"] == instance) + .unwrap(); + let estimate = row["value"][1].as_str().unwrap().parse::().unwrap(); + assert!((estimate - exact).abs() / exact <= 0.05, "{result}"); + } + eprintln!("DISTINCT_WARM {result}"); + fallback_task.abort(); +} diff --git a/docs/design_docs/README.md b/docs/design_docs/README.md index 9695976af..70c0687d2 100644 --- a/docs/design_docs/README.md +++ b/docs/design_docs/README.md @@ -26,6 +26,7 @@ under [developer docs](../developer_docs/README.md). Other designs and profiles: +- [Evidence-dependent candidate selection](evidence-dependent-candidates.md) defines evidence ownership, logical selection, physical admission, exact fallback, and current proof limits. - [ASAPQuery compatibility profile](asapquery-compatibility-profile.md) - [Shape-aware ERP](shape-aware-erp-v1.md) - [Empirical observability execution plan](empirical-o11y-execution-plan.md) diff --git a/docs/design_docs/asapplanner-integration.md b/docs/design_docs/asapplanner-integration.md index 6272905c0..2776fdb21 100644 --- a/docs/design_docs/asapplanner-integration.md +++ b/docs/design_docs/asapplanner-integration.md @@ -387,6 +387,16 @@ that output a `stored_output_id` and emits matching writer/reader bindings; see | Plan read/write bindings | State references, format, partition rules and writer ownership | | `SummaryStore` | Owns `summary_definitions` and `stored_summaries`; the latter holds committed metadata and payload together | +Costs include initialization, ingestion updates, overlapping and retained state, +transmission, storage, merges, readouts, recurring queries, and shared producer +construction once. Compare candidates over the same data and demand scope. +Missing evidence is not zero cost. The backend supplies candidate estimates +through `candidate_cost()`, preferring applicable ERP benchmarks and otherwise +using its analytical model. Unsupported estimates remain unavailable; +the backend combines unit resources, data size, recurrence, and the bound +physical DAG into complete workload costs before publication. See the +[calculation model](evidence-dependent-candidates.md#backend-owned-workload-calculation). + ## Compiler contract The compiler consumes: diff --git a/docs/design_docs/evidence-dependent-candidates.md b/docs/design_docs/evidence-dependent-candidates.md new file mode 100644 index 000000000..83c57b87b --- /dev/null +++ b/docs/design_docs/evidence-dependent-candidates.md @@ -0,0 +1,319 @@ +# Evidence-dependent candidate selection and deployment + +Status: evidence and workload costing implemented by backend PR #761, building +on the API adaptation in #768, against ASAPPlanner #455 +(`2ec3fc80`). This document defines the backend decision boundary for Planner +issue #454 and backend issue #752. It does not claim that every retained +candidate has a deployable implementation. + +## Decision and ownership + +Keep constructible candidates visible when external evidence is absent. +Separate candidate existence, logical selection, and deployment admission: +none implies the next. Otherwise the backend either loses a candidate it +could prove valid or deploys one whose guarantee has never been established. + +| Decision | Owner | Required behavior | +| --- | --- | --- | +| Construct semantic candidates | Planner | Preserve unknown guarantees; reject known-invalid evidence and impossible shapes | +| Supply external facts | Backend | Bind evidence to the query, data population, snapshot and validity period | +| Derive accuracy and select logical roots | Planner, under backend models and policy | Respect the root accuracy target; missing proof is not certification | +| Bind and admit a deployment | Backend | Verify concrete execution support and compute complete workload resource cost | + +The backend still invokes Planner's workload search and global selection. It +does not introduce a second semantic optimizer. Physical binding preserves the +selected DAG's operators, grouping, windows and dependencies; a semantic change +requires a new selection. Single-query and workload selection use the same +costed Planner search; the first-candidate helper has been removed. + +## Decision flow + +```mermaid +flowchart TD + Input[Queries, accuracy targets and optional backend evidence] --> Validate[Validate evidence scope and validity] + Validate -->|Invalid supplied evidence| Error[Reject request with reason] + Validate -->|Valid or absent evidence| Search[Planner search retains constructible candidates] + Search --> Inspect[Explain known and unknown candidate properties] + Search --> Select[Planner global selection under backend policy] + Select --> Exact[Explicit exact fallback when no certified summary is selected] + Select --> Bind[Bind selected logical DAG to concrete execution] + Bind --> Admit[Check guarantees, runtime support and computed workload cost] + Admit -->|Pass| Publish[Publish coherent physical plan] + Admit -->|Fail| Reject[Reject deployment] +``` + +Exact fallback is part of normal logical planning. A directly supplied summary +plan with missing or unknown readout guarantees fails physical compilation; +there is no promise that every compilation error retries another candidate. +An exact route must itself be available under the deployment's existing policy. + +## Evidence contract + +Evidence belongs to an exact query text, full data workload and data snapshot. +It also names its source, observation time and validity window. The backend +rejects mismatched, expired, future-dated or invalid records before selection. +Discovery without a workload quote needs an explicit data snapshot; when both +are supplied, their snapshot identities must agree. Queries with individual +certificates are isolated during selection so another query cannot borrow them. + +| Family | Facts that can justify a candidate | Missing-proof behavior | +| --- | --- | --- | +| DDSketch quantile ratio | Enforced input bounds and maximum sample count for each exact operand | Ratio remains visible with unknown propagated accuracy | +| Hydra | Shared-grid collision bound and failure probability | Missing bounds remain unknown | +| TopK | Selected lower bound, excluded upper bound and interval failure probability | No membership certificate is invented | +| Relative composition | Applicable non-negativity, cardinality and distribution facts | Unsupported propagation remains unknown | +| HLL / ERP readouts | A valid accuracy model including the required confidence guarantee | RSE or observed maximum error alone cannot certify the readout | + +Quantile domains describe an enforced source contract, not observed sample +extrema. Supplied TopK intervals must be complete and strictly separate selected +from excluded items. Omitted facts remain unknown; malformed supplied evidence +is rejected rather than treated as absent. The older TopK evidence interface +remains compatible; the scoped contract is the path for new producers. + +The backend validates the supplied record's scope and consistency. It does not +derive a source-domain proof from samples or establish the truth of a producer's +claimed contract. Producing valid external proofs remains an upstream duty. + +### Bounded classic HLL confidence + +Scoped evidence may carry `hll: {"model": +"asap-classic64-uniform-hash-linear-counting-v1", "max_distinct_per_readout": 128}`. +This declares an enforced source-domain upper bound for **each complete readout +population**, including the union of all merged panes, and the independent +uniform bucket-hash assumption. Series count, a sampled distinct count, and +ERP observed maxima do not establish this contract. The bound is supplied by +the source-contract owner; the backend validates its query/data/snapshot/time +scope, not the truth of the external source assertion. + +The model supports declared upper bounds from 1 through 4096 and precisions +4 through 18. It only certifies parameters that keep every permitted population +in classic HLL's linear-counting branch. A finite collision-arrival bound gives +failure probability for the requested relative error. Sizing searches for the +smallest supported precision meeting the same epsilon/delta contract. Explain +retains the model, population limit, hash assumption and numerical guarantee. + +The implementation is bound to the backend-local Regular HLL estimator; +collector estimators, HIP, MLE and unbounded populations are not certified by it. +Missing contracts retain generic HLL's unknown probability; infeasible targets +retain exact execution. ERP resources can still price the selected parameters, +but ERP error maxima cannot override this confidence model. Each readout's +probability is not a simultaneous guarantee for an entire dashboard. + +## Accuracy, runtime and cost remain separate + +A known accuracy guarantee does not establish executor support. The physical +compiler checks the executable DAG and runtime policy, and rejects summary +readouts whose guarantee is absent or contains unknown terms. Unknown support +must not be reported as deployment approval. + +Missing numerical cost remains unavailable, never zero. The backend implements +Planner's `candidate_cost()` directly; it does not enable uncosted legacy +selection. A cost estimate only supports logical selection. Publication requires complete backend-computed resource costs or an explicit, +applicable provider override. A cheap candidate cannot bypass +accuracy or runtime admission. + +Explain records accuracy status and symbolic guarantee, runtime support status, +cost availability, and the selection reason separately. A selected candidate +is labelled as pending backend binding. Rejected candidates retain their +reported reasons. This distinguishes missing proof, unsupported execution, +missing comparable cost and a candidate that simply lost the ranking. + +## ERP and analytical cost models + +ERP is the benchmark source for this path. The existing ERP artifact and +runtime-observation input feed both parameter planning and resource estimation; +there is no separate benchmark upload contract for candidate costs. + +```mermaid +flowchart TD + Profile[ERP benchmark artifact] --> Match[Match implementation, exact parameters and data population] + Observations[Declared distribution or validated runtime shape] --> Match + Candidate[Planner candidate with concrete parameters] --> Match + Match -->|Applicable profile| Measured[ERP resource estimate] + Match -->|No applicable profile| Analytical[Backend analytical resource estimate] + Measured --> Cost[Backend candidate_cost and source explanation] + Analytical --> Cost + Cost --> Selection[Planner logical selection] + Proof[Accuracy evidence and guarantee model] --> Selection + Selection --> Physical[Physical binding and complete workload costing] + Physical --> Admission[Deployment admission] +``` + +**Priority is applicable ERP, then analytical, then unavailable.** A larger +measured value still overrides a smaller analytical estimate. ERP matching +uses its existing implementation/runtime filters, exact sketch parameters, +minimum trial count, distribution equality or configured bounded shape match. +Catalog-scoped observations retain their existing population and freshness +validation. A profile for another parameter point or implementation cannot be +substituted. ERP v1 artifacts themselves have no per-record expiry field; do +not confuse runtime-observation freshness with a benchmark expiry guarantee. + +Cost lookup does not certify empirical accuracy: measured resource usage can +be useful even when observed error does not meet a target or cannot establish +its failure probability. The cost path reuses ERP profile matching without an +empirical-error threshold; the accuracy path separately checks the requested +guarantee. In particular, an explicit confidence target must not erase ERP +resource measurements merely because ERP v1 cannot prove that confidence. +Existing empirical-only accuracy policy still applies to accuracy decisions. + +The first backend model, `backend_state_footprint_v1`, estimates **retained +state bytes per partition** for local candidate selection. It uses ERP's +`memory_bytes` when applicable; otherwise it reuses the backend's existing +analytical retained-state formulas: matrix dimensions and heap capacity, +KLL capacity, HLL registers, fixed accumulator allowance and the current +DDSketch allowance. These are estimates, not measured limits or a complete +workload cost. Reachable shared state nodes are counted once. For multiple +observed populations the ERP proxy uses the largest matched partition, without +pooling the populations. Shared-grid families need their own applicable model; +an independent sketch profile is not a Hydra-grid measurement. + +The estimate excludes population counts, pane multiplicity, CPU, transmission +and other deployment costs. An applicable ERP profile also ranks sketch-family +candidates using the same byte estimates, with analytical estimates for the +unmeasured peers. Without an applicable profile, existing family preference +order remains the fallback policy. +It does not mix ERP CPU seconds with analytical bytes or claim to minimize +complete workload cost. Raw rewrites and exact compositions have no state-byte +estimate here; exact composition retains its separate measured recurring-cost +model. Unsupported shapes remain uncosted. + +Explain attaches the model, unit, source (`erp`, `analytical`, or `mixed`) and +ERP record IDs to each available estimate. Final deployment selection combines these unit resources with physical demand +as described below. A profile alone neither prices the workload nor approves +publication. + +Acceptance covers ERP precedence even when measured cost is higher, matching +parameters/implementation/population, insufficient trials, invalid measurements, +analytical fallback, unavailable shapes, explain provenance, and independent +accuracy and publication gates. + +## Backend-owned workload calculation + +Both backend-local startup and the PromQL/MetricsQL compile-and-publish API +compute workload costs when `workload_cost_evidence` is absent. Callers supply +queries, recurrence, data facts, deployment capabilities and optional ERP; they +do not need to manufacture complete candidate quotes. The backend binds the +bounded candidate inventory, constructs its coverage manifest, calculates every +component, and selects the lowest comparable total among feasible candidates. +The logical state-footprint proxy above remains a separate early selection +model; physical selection compares only the enumerated candidates, not every +possible logical DAG or window implementation. + +```mermaid +flowchart LR + ERP[Applicable ERP unit resources] --> Units[ERP first; analytical fallback] + Data[Fresh rate, cardinality, cadence, volume] --> Demand[Physical workload quantities] + Queries[Query frequency and common horizon] --> Demand + Plan[Bound DAG, shared state, panes, locations, transport] --> Demand + Units --> Compute[CPU seconds, memory byte-seconds, network bytes] + Demand --> Compute + Compute --> Compare[Common weights and complete component coverage] + Compare --> Select[Lowest-cost feasible physical candidate] +``` + +`backend-workload-resources-v1` is an explicit analytical reference model, not a +calibrated prediction or a monetary quote. It computes a resource vector before +weighting it: `cost = CPU_seconds + 1e-9 * memory_byte_seconds + 1e-8 * network_bytes`. +These fixed versioned weights express a default tradeoff; a deployment can still +supply complete calibrated provider quotes as an explicit override. Override +quotes are validated as one model across the inventory; missing/invalid quotes +are not silently patched with analytical values in unrelated units. + +For horizon H, input rate R, query interval I seconds and partition estimate P: + +| Component | Backend calculation | +| --- | --- | +| Source | R × H samples per distinct source/location; exact sources also retain raw samples for the required lookback plus series metadata | +| State initialization/retirement | P × (initial retained states + state rotations during H); allocation charges estimated bytes, retirement charges state count | +| State updates | R × H × overlapping full windows; panes receive each input once; rollup levels add their child-to-parent merges | +| State residency | P × retained physical states × unit state bytes × H | +| Transport | Rule emission/checkpoint count × partitions × retained states × estimated encoded bytes, plus serialization, decoding and receiving merges | +| Summary reads | H / I × P × panes read; decode each pane, merge additional panes, then charge the selected state's query/readout unit cost | +| Other query operators | Traverse each reachable physical node once per evaluation, using propagated row estimates and an explicit analytical operation model | +| Native exact subtrees | Source sample count for the lookback or full data volume, syntax complexity and a sorting allowance; charge each subtree's returned data separately from final output | +| Result | Result row estimate × row bytes × H / I, including serialization and delivery | + +ERP supplies state memory and per-update, per-merge and per-query CPU seconds. +The backend uses the actual workload quantities, not ERP's example invocation +counts. Applicability checks are the same implementation/parameter/runtime/ +population/shape/trial checks used by candidate costs. Multiple matched +population profiles use the maximum of each resource dimension per partition. +ERP measurements override analytical estimates even when more expensive. They +do not establish accuracy guarantees or runtime availability. + +Unmeasured dimensions use versioned analytical assumptions: 1e-7 CPU seconds per +item, 1e-9 CPU seconds per byte, 24 bytes per raw sample, and a 256-byte series/ +result-row allowance. Sketch update work scales with log2(state bytes / 16), +while merge/read work scales with state bytes. Native exact work uses +`(samples + 1) × log2(max(samples, 2)) × canonical syntax object count × item CPU`. +This is a transparent complexity proxy, not a benchmark of Prometheus; local +arithmetic/reduction/sort and joins have their own row-based estimates. + +The current data contract has workload-level facts, not a complete per-source +histogram. The model explicitly replicates the workload rate/cardinality to +each distinct source and collector rather than claiming known selectivities. +Grouped states use input cardinality as a group-count estimate. Without a fresh +cardinality, it estimates `ceil(R × scrape_interval)` under the stated assumption +of one sample per active series per scrape; a zero rate cannot establish an +unknown population. At-rest/full-history costing requires fresh ingestion volume. +Unknown rate/cadence, unsupported operators, invalid values and overflow make a +candidate unavailable. No partial total is admitted. + +Shared state and source upkeep are charged once per physical identity/location; +additional consumers add reads and outputs. For distributed plans, receiving +backend merges are charged in transport rather than again as raw updates. +Full-window overlap and pane retention are taken from the compiled layout; +rollup reads conservatively use base panes. Delta transport conservatively uses +full payload size and checkpoints; there is no unmeasured compression discount. +All recurring work uses the same horizon and original query demand. Existing +window implementation/lifecycle inputs still determine which physical layouts +are bound; their opaque weighted costs are not mixed with this resource model. + +Each candidate's `automatic_cost` report records the input data, decision time, +model/weights, assumptions, component resource vectors, workload multipliers and +applicable ERP record IDs. `component_costs` is the weighted projection of that +report, and covers exactly the existing manifest. Invalid provider overrides +remain explicit errors. Accuracy proofs, compiler capability checks and runtime +readiness/fallback policy remain independent admission requirements. + +Acceptance includes quote-free startup and HTTP compilation, ERP precedence and +implementation mismatch, frequency/data-size scaling, shared-state deduplication, +expired facts, overflow, complete coverage and provider-override compatibility. + +## Example and acceptance behavior + +For `quantile_over_time(0.9, data[5m]) / quantile_over_time(0.5, data[5m])`: + +1. Without operand-domain evidence, the DDSketch ratio remains inspectable but + has unknown accuracy. Normal planning preserves exact execution. +2. With valid, scoped operand contracts, Planner can derive the ratio guarantee + and check the explicit root target. Valid evidence alone does not guarantee + that the target is met or that this candidate wins selection. +3. A selected candidate still needs physical support and complete backend-computed + workload costs before publication. A stale or cross-query certificate rejects the request. + +Cross-family acceptance includes absent, partial, valid, invalid and stale +evidence, an explicit root accuracy target, unavailable cost, and unsupported +runtime operations. Explain must never call an uncertified candidate approved; +direct physical submission must not bypass the guarantee check. + +The current conservative policy changes behavior for HLL and ERP v1: relative +standard error and benchmark maximum error lack a calibrated tail probability. +Those paths use exact execution when no independent valid guarantee exists. +Re-enabling them requires an appropriate proof model, not merely more benchmark +samples or a favorable shape-match score. + +Related: [integration architecture](asapplanner-integration.md), +[shape-aware ERP](shape-aware-erp-v1.md), +[Planner evidence contract](https://github.com/ProjectASAP/ASAPPlanner/blob/2ec3fc80caa922c8e1f33aa05f60b7d787e73257/docs/design_docs/architecture/evidence-dependent-candidates.md). + +## Time precision at admission + +Source cadence, query lookback, ranges and offsets retain integral millisecond +precision through workload lowering and query publication. A 100 ms source is +not rejected or rewritten to one second before costing. Existing pane layout +contracts still express storage sizes in seconds; rounding a storage sizing +bound must not change the query's read interval. A temporal producer that cannot +implement a fractional range remains unsupported for that binding, while exact +execution preserves the original range. Level-3 acceptance uses the actual +replay cadence and still requires a local executable plan. diff --git a/docs/design_docs/shape-aware-erp-v1.md b/docs/design_docs/shape-aware-erp-v1.md index 5d04d99af..0f0a7fccf 100644 --- a/docs/design_docs/shape-aware-erp-v1.md +++ b/docs/design_docs/shape-aware-erp-v1.md @@ -22,8 +22,12 @@ than publishing a biased partial snapshot. Profiles with too few benchmark events, poor fit, ambiguous confidence, or excessive cardinality/parameter distance are misses. -On a hit, empirical parameters and measured atomic costs are used. On a miss, -malformed evidence, or drift, Hybrid mode retains the theoretical parameters; +On a hit, empirical parameters and measured atomic costs inform planning. +ERP resource precedence, analytical fallback, accuracy certification and +deployment admission follow the +[evidence-dependent candidate design](evidence-dependent-candidates.md#erp-and-analytical-cost-models). + +On a miss, malformed evidence, or drift, Hybrid mode retains the theoretical parameters; if the runtime cannot deploy them or they exceed its memory limit, compilation chooses exact execution. Empirical-only mode fails closed. diff --git a/docs/evaluation/e2e-physical-dag.md b/docs/evaluation/e2e-physical-dag.md index 0666cefbf..6876bead4 100644 --- a/docs/evaluation/e2e-physical-dag.md +++ b/docs/evaluation/e2e-physical-dag.md @@ -39,9 +39,10 @@ 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 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. +is not an empirical-ERP benchmark. For deployment, the backend automatically computes workload costs from ERP or +analytical unit resources and physical demand. Optional calibrated overrides use +the [cost evidence workflow](../examples/workload-cost-evidence.md). Do not relabel +analytical 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): @@ -73,12 +74,11 @@ dot -Tsvg target/physical-dag-inspection/selected.dot \ -o target/physical-dag-inspection/selected.svg ``` -`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. +`compile_workload_artifact` calls the same snapshot compiler used by startup. +Set `ASAPQUERY_PLANNING_SNAPSHOT` to the workload snapshot. Without an explicit +provider override it uses automatic ERP/analytical workload costing. The output +includes the selected plan, logical selection trace, resource assumptions and +complete cost comparison. Both PromQL and MetricsQL use this path. The compiler derives sibling plans from the selected post-ASAP DAG: diff --git a/docs/examples/workload-cost-evidence.md b/docs/examples/workload-cost-evidence.md index 64e60498c..7adaefcc7 100644 --- a/docs/examples/workload-cost-evidence.md +++ b/docs/examples/workload-cost-evidence.md @@ -1,10 +1,16 @@ # Complete workload cost evidence 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. +Deployment automatically calculates workload costs when `workload_cost_evidence` +is absent, using applicable ERP profiles and analytical estimates together with +the workload's data facts, query frequency and physical layout. Inspect +`cost_comparison` for candidate totals, resource breakdowns and assumptions. +The checked-in examples can exercise this path at their recorded decision time; +use current data/capability inputs for a live deployment. + +The following workflow is an **optional provider override** for deployments with +calibrated complete quotes. See the [automatic calculation design](../design_docs/evidence-dependent-candidates.md#backend-owned-workload-calculation) +for the default path. ## Workflow @@ -65,7 +71,9 @@ can change which bound workload is committed, not rewrite its semantics. All quotes use one provider model's common cost units. A horizon quote includes the complete stated partition's work, data volume/cardinality, maintained -groups and retention. Do not reuse a global ingestion rate as a per-metric rate. +groups and retention. A calibrated provider should supply per-source demand rather than infer it from +a global rate. The automatic reference model discloses conservative replication +when only workload-level facts are available. Per-evaluation quotes exclude upkeep already charged in horizon components. Sunk infrastructure may explicitly cost zero under the provider's documented decision boundary; unknown costs may not.