diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 460b5e531..e06f61776 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -2482,17 +2482,18 @@ pub fn select_workload_roots_with_erp( // confidence requirements through theoretical/exact fallback. let scoped_erp = erp.map(|policy| { let mut policy = policy.clone(); - if !matches!(accuracy, AccuracyTarget::Epsilon(_)) - || policy.error_metric != "max_rank_err" - { + if !matches!(accuracy, AccuracyTarget::Epsilon(_)) { policy.artifact.records.clear(); } // A benchmark of a different KLL implementation is not evidence // for the collector's sketchlib KLL, even with the same k. - policy - .artifact - .records - .retain(|row| row.sketch == "kll-percall" && row.implementation == "lib"); + 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") + }); policy }); let erp = scoped_erp.as_ref(); @@ -2542,16 +2543,39 @@ fn requires_exact_erp_fallback( accuracy: &AccuracyTarget, erp: &super::erp::ErpPlanningInput, ) -> bool { - fn walk(node: &SummaryNode, out: &mut Vec<(SketchAlgorithm, SketchParams)>) { + fn walk( + node: &SummaryNode, + out: &mut Vec<( + SketchAlgorithm, + SketchParams, + Option, + )>, + ) { match &node.expr { - SummaryExpr::SummaryAgg { family, child, .. } => { - if let SummaryFamilyType::Sketch(kind, _) = family { - out.push((kind.algorithm().clone(), kind.params().clone())); - } + SummaryExpr::SummaryAgg { child, .. } => { walk(child, out); } - SummaryExpr::SummaryEstimate { summary_input, .. } - | SummaryExpr::SummaryDelete { summary_input, .. } + SummaryExpr::SummaryEstimate { + summary_input, + query, + .. + } => { + for field in &summary_input.schema.fields { + if node.guarantee.as_ref().is_some_and(|g| g.is_exact()) { + continue; + } + let SummaryFamilyType::Sketch(kind, _) = &field.dtype else { + continue; + }; + out.push(( + kind.algorithm().clone(), + kind.params().clone(), + super::erp::ReadoutEvidence::for_query(kind.algorithm(), query), + )); + } + walk(summary_input, out); + } + SummaryExpr::SummaryDelete { summary_input, .. } | SummaryExpr::ValueOperation { child: summary_input, .. @@ -2590,9 +2614,13 @@ fn requires_exact_erp_fallback( }; let mut sketches = Vec::new(); walk(node, &mut sketches); - sketches.into_iter().any(|(algorithm, params)| { + sketches.into_iter().any(|(algorithm, params, readout)| { + let decision = match readout { + Some(readout) => erp.select_readout(algorithm, readout, max_error, params), + None => erp.select(algorithm, max_error, params), + }; matches!( - erp.select(algorithm, max_error, params), + decision, super::erp::ErpParameterDecision::ExactFallback { .. } ) }) diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs index 0dfe4c871..a638e38a8 100644 --- a/control_plane/src/physical/erp.rs +++ b/control_plane/src/physical/erp.rs @@ -165,6 +165,75 @@ fn fit_zipf_exponent(descending_counts: &[u64]) -> f64 { } } +/// The measurement units are part of the readout contract, not a property of +/// the shared sketch state. These keys must be supplied by the benchmark. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ReadoutEvidence { + QuantileRank, + DistinctRelative, + FrequencyL2Relative, + EntropyAbsoluteBits, +} + +impl ReadoutEvidence { + pub(crate) fn for_intent( + algorithm: &SketchAlgorithm, + intent: &planner_types::pre_asap::AggIntent, + ) -> Option { + use planner_types::pre_asap::AggIntent; + match (algorithm, intent) { + (SketchAlgorithm::Kll, AggIntent::Quantile { .. }) => Some(Self::QuantileRank), + (SketchAlgorithm::Hll | SketchAlgorithm::UnivMon, AggIntent::Cardinality { .. }) => { + Some(Self::DistinctRelative) + } + (SketchAlgorithm::UnivMon, AggIntent::FrequencyL2 { .. }) => { + Some(Self::FrequencyL2Relative) + } + (SketchAlgorithm::UnivMon, AggIntent::FrequencyEntropy { .. }) => { + Some(Self::EntropyAbsoluteBits) + } + _ => None, + } + } + + pub(crate) fn for_query( + algorithm: &SketchAlgorithm, + query: &planner_types::post_asap::SketchQuery, + ) -> Option { + use planner_types::post_asap::SketchQuery; + match (algorithm, query) { + (SketchAlgorithm::Kll, SketchQuery::Quantile { .. }) => Some(Self::QuantileRank), + (SketchAlgorithm::Hll | SketchAlgorithm::UnivMon, SketchQuery::Cardinality) => { + Some(Self::DistinctRelative) + } + (SketchAlgorithm::UnivMon, SketchQuery::FrequencyL2) => Some(Self::FrequencyL2Relative), + (SketchAlgorithm::UnivMon, SketchQuery::FrequencyEntropy) => { + Some(Self::EntropyAbsoluteBits) + } + _ => None, + } + } + + fn metric_key(self) -> &'static str { + match self { + Self::QuantileRank => "max_rank_err", + Self::DistinctRelative => "max_cardinality_relative_error", + Self::FrequencyL2Relative => "max_frequency_l2_relative_error", + Self::EntropyAbsoluteBits => "max_frequency_entropy_absolute_bits_error", + } + } + + fn metric(self) -> planner_types::post_asap::ErrorMetric { + use planner_types::post_asap::ErrorMetric; + match self { + Self::QuantileRank => ErrorMetric::Rank, + Self::DistinctRelative => ErrorMetric::Cardinality, + Self::FrequencyL2Relative => ErrorMetric::RelativeValue, + Self::EntropyAbsoluteBits => ErrorMetric::AbsoluteValue, + } + } +} + /// ERP v1 measures error magnitudes, not tail probabilities. Only an explicit /// epsilon-only request may use these observations as its accuracy contract. pub(crate) struct ErpAccuracyModel<'a> { @@ -185,51 +254,64 @@ impl asap_aware_mapping::AccuracyModel for ErpAccuracyModel<'_> { query: &planner_types::post_asap::SketchQuery, ) -> Option { use planner_types::post_asap::*; - let mut guarantee = - asap_aware_mapping::DefaultAccuracyModel.local_guarantee(family, query)?; - if let (Some(policy), SummaryFamilyType::Sketch(kind, _)) = (self.policy, family) { - let decision = policy.select( + let theoretical = asap_aware_mapping::DefaultAccuracyModel.local_guarantee(family, query); + let (Some(policy), SummaryFamilyType::Sketch(kind, _)) = (self.policy, family) else { + return theoretical; + }; + let Some(readout) = ReadoutEvidence::for_query(kind.algorithm(), query) else { + // In particular, UnivMon's total unit count is exact without + // empirical error evidence. No unsupported readout gets a bound. + if theoretical.as_ref().is_some_and(ResultGuarantee::is_exact) { + return theoretical; + } + return match policy.select( kind.algorithm().clone(), self.max_error, kind.params().clone(), - ); - if matches!(decision, ErpParameterDecision::ExactFallback { .. }) { - return None; - } - if let ErpParameterDecision::Empirical { + ) { + ErpParameterDecision::ExactFallback { .. } => None, + _ => theoretical, + }; + }; + match policy.evidence_for_readout( + kind.algorithm().clone(), + readout, + self.max_error, + kind.params(), + ) { + ErpParameterDecision::ExactFallback { .. } => None, + ErpParameterDecision::TheoreticalFallback { .. } => theoretical, + ErpParameterDecision::Empirical { params, record_id, observed_error, .. - } = decision - { - if ¶ms == kind.params() { - // This is the only benchmark-to-query metric mapping currently - // validated end to end. Means and value errors are not rank bounds. - if guarantee.metric != ErrorMetric::Rank - || policy.error_metric != "max_rank_err" - { - return None; - } - guarantee.bound = BoundExpr::Constant { + } => { + if ¶ms != kind.params() { + return None; + } + Some(ResultGuarantee { + metric: readout.metric(), + bound: BoundExpr::Constant { value: observed_error, - }; - guarantee.failure_probability = ProbabilityExpr::Unknown { + }, + failure_probability: ProbabilityExpr::Unknown { statistic: "erp_v1_has_no_failure_probability_evidence".into(), - }; - guarantee.provenance = vec![GuaranteeSource::SketchReadout { + }, + provenance: vec![GuaranteeSource::SketchReadout { algorithm: format!("{:?}", kind.algorithm()), contract: format!( - "erp_v1_empirical:{}:{}", - policy.artifact.producer_version, record_id + "erp_v1_empirical:{}:{}:{}", + policy.artifact.producer_version, + record_id, + readout.metric_key() ), params: serde_json::to_value(¶ms).ok()?, query: format!("{query:?}"), - }]; - } + }], + }) } } - Some(guarantee) } fn propagate( @@ -388,6 +470,51 @@ impl ErpPlanningInput { algorithm: SketchAlgorithm, max_error: f64, theoretical: SketchParams, + ) -> ErpParameterDecision { + self.select_metric(algorithm, &self.error_metric, max_error, theoretical, None) + } + + pub(crate) fn select_readout( + &self, + algorithm: SketchAlgorithm, + readout: ReadoutEvidence, + max_error: f64, + theoretical: SketchParams, + ) -> ErpParameterDecision { + self.select_metric( + algorithm, + readout.metric_key(), + max_error, + theoretical, + None, + ) + } + + /// Validate an existing state's readout independently of which new state + /// would be cheapest to allocate for this one consumer. + fn evidence_for_readout( + &self, + algorithm: SketchAlgorithm, + readout: ReadoutEvidence, + max_error: f64, + actual: &SketchParams, + ) -> ErpParameterDecision { + self.select_metric( + algorithm, + readout.metric_key(), + max_error, + actual.clone(), + Some(actual), + ) + } + + fn select_metric( + &self, + algorithm: SketchAlgorithm, + error_metric: &str, + max_error: f64, + theoretical: SketchParams, + required_params: Option<&SketchParams>, ) -> ErpParameterDecision { // Runtime admissibility belongs before ranking: an unusable cheap // profile must not hide a more expensive executable alternative. @@ -395,8 +522,12 @@ impl ErpPlanningInput { artifact.records.retain(|row| { sketch_name_matches(&row.sketch, &algorithm) && parse_params(&algorithm, &row.parameters).is_some_and(|params| { - self.runtime - .supports(&algorithm, ¶ms, Some(row.resources.memory_bytes)) + required_params.is_none_or(|required| required == ¶ms) + && self.runtime.supports( + &algorithm, + ¶ms, + Some(row.resources.memory_bytes), + ) }) }); let allowed_sketches = artifact @@ -408,7 +539,7 @@ impl ErpPlanningInput { distribution: self.distribution.clone(), implementation: self.implementation.clone(), allowed_sketches, - error_metric: self.error_metric.clone(), + error_metric: error_metric.to_owned(), max_error, min_trials: self.min_trials, expected_updates: self.expected_updates, @@ -528,6 +659,7 @@ fn sketch_name_matches(name: &str, algorithm: &SketchAlgorithm) -> bool { SketchAlgorithm::Hll => name.starts_with("hll") || name.starts_with("hyperloglog"), SketchAlgorithm::Kll => name.starts_with("kll"), SketchAlgorithm::DDSketch => name.starts_with("ddsketch"), + SketchAlgorithm::UnivMon => name == "univmon", _ => false, } } @@ -582,6 +714,12 @@ fn parse_params(algorithm: &SketchAlgorithm, parameters: &Value) -> Option SketchParams::Kll { k: u32_param(parameters, &["k"])?, }, + SketchAlgorithm::UnivMon => SketchParams::UnivMon { + heap_size: u32_param(parameters, &["heap_size"])?, + sketch_rows: u32_param(parameters, &["sketch_rows"])?, + sketch_cols: u32_param(parameters, &["sketch_cols"])?, + layers: u8::try_from(u32_param(parameters, &["layers"])?).ok()?, + }, SketchAlgorithm::DDSketch => SketchParams::DDSketch { alpha: number(parameters, &["alpha", "relative_accuracy"])?, }, @@ -611,6 +749,24 @@ fn valid_runtime_params(algorithm: &SketchAlgorithm, params: &SketchParams) -> b heap_size, }, ) => *width >= 2 && width.is_power_of_two() && *depth >= 1 && *heap_size >= 1, + ( + SketchAlgorithm::UnivMon, + SketchParams::UnivMon { + heap_size, + sketch_rows, + sketch_cols, + layers, + }, + ) => { + *heap_size > 0 + && *sketch_cols > 0 + && (1..=20).contains(sketch_rows) + && (1..=64).contains(layers) + && sketch_rows + .checked_mul(*sketch_cols) + .and_then(|n| n.checked_mul(u32::from(*layers))) + .is_some() + } (SketchAlgorithm::Hll, SketchParams::Hll { precision }) => (4..=18).contains(precision), (SketchAlgorithm::Kll, SketchParams::Kll { k }) => (8..=65_535).contains(k), (SketchAlgorithm::DDSketch, SketchParams::DDSketch { alpha }) => { @@ -684,6 +840,176 @@ mod tests { } } + /// Structural evidence fixture, not measured calibration data. + #[test] + fn existing_univmon_readout_uses_its_own_evidence_despite_cheaper_sibling() { + use asap_aware_mapping::AccuracyModel; + use planner_types::post_asap::*; + let mut policy = input(ErpAccuracyMode::Empirical); + let small = &mut policy.artifact.records[0]; + small.sketch = "univmon".into(); + small.parameters = + serde_json::json!({"heap_size":32,"sketch_rows":5,"sketch_cols":128,"layers":4}); + small.error_metrics = + BTreeMap::from([("max_frequency_entropy_absolute_bits_error".into(), 0.1)]); + let mut large = small.clone(); + large.id = "larger-shared-state".into(); + large.parameters["heap_size"] = 128.into(); + large.resources.memory_bytes *= 2.0; + large + .error_metrics + .insert("max_frequency_entropy_absolute_bits_error".into(), 0.02); + policy.artifact.records.push(large); + let small_params = SketchParams::UnivMon { + heap_size: 32, + sketch_rows: 5, + sketch_cols: 128, + layers: 4, + }; + let actual = SketchParams::UnivMon { + heap_size: 128, + sketch_rows: 5, + sketch_cols: 128, + layers: 4, + }; + assert_eq!( + policy + .select_readout( + SketchAlgorithm::UnivMon, + ReadoutEvidence::EntropyAbsoluteBits, + 0.2, + actual.clone() + ) + .params(), + Some(&small_params) + ); + let family = SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::UnivMon, actual), + GroupingStrategy::PerSubpopulationInstance, + ); + let model = ErpAccuracyModel { + policy: Some(&policy), + max_error: 0.2, + }; + let guarantee = model + .local_guarantee(&family, &SketchQuery::FrequencyEntropy) + .unwrap(); + assert_eq!(guarantee.bound, BoundExpr::Constant { value: 0.02 }); + assert!(matches!( + guarantee.failure_probability, + ProbabilityExpr::Unknown { .. } + )); + policy.artifact.records[1].error_metrics.clear(); + assert!(ErpAccuracyModel { + policy: Some(&policy), + max_error: 0.2 + } + .local_guarantee(&family, &SketchQuery::FrequencyEntropy) + .is_none()); + } + + #[test] + fn hll_cardinality_uses_measured_relative_error_not_rse() { + use asap_aware_mapping::AccuracyModel; + use planner_types::post_asap::*; + let mut policy = input(ErpAccuracyMode::Hybrid); + let row = &mut policy.artifact.records[0]; + row.sketch = "hll".into(); + row.parameters = serde_json::json!({"precision":12}); + row.error_metrics = BTreeMap::from([("max_cardinality_relative_error".into(), 0.04)]); + let family = SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Hll, SketchParams::Hll { precision: 12 }), + GroupingStrategy::PerSubpopulationInstance, + ); + let model = ErpAccuracyModel { + policy: Some(&policy), + max_error: 0.05, + }; + let guarantee = model + .local_guarantee(&family, &SketchQuery::Cardinality) + .unwrap(); + assert_eq!(guarantee.metric, ErrorMetric::Cardinality); + assert_eq!(guarantee.bound, BoundExpr::Constant { value: 0.04 }); + assert!(matches!( + guarantee.failure_probability, + ProbabilityExpr::Unknown { .. } + )); + assert_eq!( + ReadoutEvidence::for_query(&SketchAlgorithm::Hll, &SketchQuery::FrequencyEntropy), + None + ); + } + + /// Contract fixture only; the process test measures real sketch errors. + #[test] + fn readout_evidence_keeps_units_and_missing_metrics_fail_closed() { + use asap_aware_mapping::AccuracyModel; + use planner_types::post_asap::*; + let mut policy = input(ErpAccuracyMode::Hybrid); + let row = &mut policy.artifact.records[0]; + row.sketch = "univmon".into(); + row.parameters = + serde_json::json!({"heap_size": 32, "sketch_rows": 5, "sketch_cols": 128, "layers": 4}); + row.error_metrics = BTreeMap::from([ + ("max_cardinality_relative_error".into(), 0.03), + ("max_frequency_l2_relative_error".into(), 0.02), + ("max_frequency_entropy_absolute_bits_error".into(), 0.1), + ]); + let family = SummaryFamilyType::Sketch( + SketchKind::new( + SketchAlgorithm::UnivMon, + SketchParams::UnivMon { + heap_size: 32, + sketch_rows: 5, + sketch_cols: 128, + layers: 4, + }, + ), + GroupingStrategy::PerSubpopulationInstance, + ); + let model = ErpAccuracyModel { + policy: Some(&policy), + max_error: 0.2, + }; + for (query, metric, bound) in [ + (SketchQuery::Cardinality, ErrorMetric::Cardinality, 0.03), + (SketchQuery::FrequencyL2, ErrorMetric::RelativeValue, 0.02), + ( + SketchQuery::FrequencyEntropy, + ErrorMetric::AbsoluteValue, + 0.1, + ), + ] { + let guarantee = model.local_guarantee(&family, &query).unwrap(); + assert_eq!(guarantee.metric, metric); + assert_eq!(guarantee.bound, BoundExpr::Constant { value: bound }); + assert!(matches!( + guarantee.failure_probability, + ProbabilityExpr::Unknown { .. } + )); + assert!(!model.satisfies( + &guarantee, + &crate::types_v2::AccuracyTarget::EpsilonDelta { + epsilon: 0.2, + delta: 0.01 + } + )); + } + policy.artifact.records[0] + .error_metrics + .remove("max_frequency_entropy_absolute_bits_error"); + let model = ErpAccuracyModel { + policy: Some(&policy), + max_error: 0.2, + }; + assert!(model + .local_guarantee(&family, &SketchQuery::FrequencyEntropy) + .is_none()); + assert!(model + .local_guarantee(&family, &SketchQuery::FrequencyL2) + .is_some()); + } + /// A matching benchmark context may reduce CMS state below theory. #[test] fn matching_profile_selects_empirical_parameters() { diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs index cad7144df..9fc8e9230 100644 --- a/control_plane/src/physical/post_asap/cost_model.rs +++ b/control_plane/src/physical/post_asap/cost_model.rs @@ -457,7 +457,9 @@ fn intent_accuracy(intent: &AggIntent) -> AccuracyTarget { match intent { AggIntent::Quantile { accuracy, .. } | AggIntent::Cardinality { accuracy, .. } - | AggIntent::TopK { accuracy, .. } => accuracy.clone(), + | AggIntent::TopK { accuracy, .. } + | AggIntent::FrequencyL2 { accuracy, .. } + | AggIntent::FrequencyEntropy { accuracy, .. } => accuracy.clone(), AggIntent::Count { accuracy } => accuracy.clone(), _ => AccuracyTarget::Exact, } @@ -675,7 +677,16 @@ impl CostModel for ControlPlaneCostModel { (eps, params) } }; - match self.erp_parameter_decision(kind, max_error, theoretical.clone()) { + let decision = match ( + &self.erp, + super::super::erp::ReadoutEvidence::for_intent(&kind, intent), + ) { + (Some(policy), Some(readout)) => { + Some(policy.select_readout(kind, readout, max_error, theoretical.clone())) + } + _ => self.erp_parameter_decision(kind, max_error, theoretical.clone()), + }; + match decision { Some(ErpParameterDecision::Empirical { params, record_id, diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 5b5e1e8a5..c5f0b2830 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -1658,3 +1658,6 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() assert!(metrics.contains("asap_remote_write_duplicates_total 33")); assert!(metrics.contains("asap_remote_write_rejected_requests_total 1")); } + +#[path = "support/univmon_erp_process.rs"] +mod univmon_erp_process; diff --git a/data_plane/tests/support/univmon_erp_process.rs b/data_plane/tests/support/univmon_erp_process.rs new file mode 100644 index 000000000..a37bc601e --- /dev/null +++ b/data_plane/tests/support/univmon_erp_process.rs @@ -0,0 +1,284 @@ +use super::*; +use control_plane::physical::{compiler::BackendLocalPlanningSnapshot, erp::ErpShapeObserver}; +use data_plane::precompute_engine::operators::univmon_accumulator::UnivMonAccumulator; +use data_plane::storage_engines::types::{AggregateCore, SerializableToSink}; + +fn values(offset: usize) -> Vec { + (1..=128) + .flat_map(|key| std::iter::repeat_n((key + offset) as f64, 256 / key)) + .collect() +} + +fn truth(raw: &[f64]) -> [f64; 3] { + let mut counts = std::collections::HashMap::::new(); + for value in raw { + *counts.entry(value.to_bits()).or_default() += 1; + } + let l2 = counts + .values() + .map(|n| (*n as f64).powi(2)) + .sum::() + .sqrt(); + let entropy = counts + .values() + .map(|n| { + let p = *n as f64 / raw.len() as f64; + -p * p.log2() + }) + .sum(); + [counts.len() as f64, l2, entropy] +} + +fn measured_artifact() -> Value { + let mut records = Vec::new(); + for (heap, cols) in [(8, 128), (128, 1024)] { + let mut errors = [0.0f64; 3]; + let mut bytes = 0; + // Independent key populations exercise fixed implementation hashes. + // These are empirical trials, not a claimed tail-probability bound. + for trial in 0..10 { + let raw = values(trial * 1000); + let exact = truth(&raw); + let mut panes = [ + UnivMonAccumulator::new(heap, 5, cols, 4).unwrap(), + UnivMonAccumulator::new(heap, 5, cols, 4).unwrap(), + ]; + for (i, value) in raw.iter().enumerate() { + panes[i % 2].insert_sample(*value).unwrap(); + } + let other = panes[1].clone(); + panes[0].merge_in_place(&other).unwrap(); + bytes = bytes.max(panes[0].serialize_to_bytes().len()); + for (i, stat) in [ + asap_types::Statistic::Cardinality, + asap_types::Statistic::FrequencyL2, + asap_types::Statistic::FrequencyEntropy, + ] + .into_iter() + .enumerate() + { + let estimate = panes[0] + .query_statistic(stat, &None, &Default::default()) + .unwrap(); + assert!(estimate.is_finite()); + let error = (estimate - exact[i]).abs() / if i == 2 { 1.0 } else { exact[i] }; + errors[i] = errors[i].max(error); + } + } + records.push(serde_json::json!({ + "id": format!("univmon-unit-frequency-h{heap}-c{cols}"), + "sketch": "univmon", "implementation": "asap-sketchlib-univmon-standard-v1", + "parameters": {"heap_size": heap, "sketch_rows": 5, "sketch_cols": cols, "layers": 4}, + "trials": 10, + "distribution": {"erp_shape": {"family": "zipf", "parameters": {"exponent": 1.0}, "cardinality": 128, "benchmark_events": values(0).len()}}, + "error_metrics": { + "max_cardinality_relative_error": errors[0], + "max_frequency_l2_relative_error": errors[1], + "max_frequency_entropy_absolute_bits_error": errors[2] + }, + "resources": {"memory_bytes": bytes, "update_cpu_seconds": 0.0, "query_cpu_seconds": 0.0, "merge_cpu_seconds": 0.0} + })); + } + serde_json::json!({"schema_version": 1, "producer_version": "backend-univmon-standard-unit-frequency-two-pane-error-fixture-cpu-not-measured", "records": records}) +} + +#[tokio::test] +async fn measured_readout_evidence_selects_and_executes_univmon() { + let artifact = measured_artifact(); + eprintln!("UNIVMON_MEASURED {artifact}"); + let raw = values(100_000); + let exact = truth(&raw); + let mut observer = ErpShapeObserver::new(128).unwrap(); + for (i, value) in raw.iter().enumerate() { + observer.observe(&value.to_string(), i / 100).unwrap(); + } + let observation = observer.snapshot().unwrap(); + let queries = [ + "distinct_over_time(erp_frequency[5s])", + "l2_over_time(erp_frequency[5s])", + "entropy_over_time(erp_frequency[5s])", + ]; + let mut fixture: Value = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let template = fixture["query_workload"]["repeating_queries"][3].clone(); + fixture["query_workload"]["repeating_queries"] = queries + .iter() + .map(|query| { + let mut entry = template.clone(); + entry["query"] = (*query).into(); + entry["requirements"]["accuracy"] = serde_json::json!({"explicit": {"Epsilon": 0.2}}); + entry + }) + .collect::>() + .into(); + fixture["implementation"]["erp"] = serde_json::json!({ + "distribution": {"workload": {"external": {"dataset": "held-out-frequency-population"}}}, + "artifact": artifact, "implementation": null, "error_metric": "readout_specific", + "min_trials": 10, "expected_updates": raw.len(), "expected_queries": 10.0, + "expected_merges": 1.0, "retention_seconds": 60.0, "cpu_weight": 0.0, + "byte_second_weight": 1e-9, "mode": "hybrid", "observed_shape": observation.observation, + "shape_match": {"minimum_benchmark_events": 1000, "max_log2_cardinality_distance": 0.0, + "max_parameter_distance": 0.2, "max_goodness_of_fit": 0.2, + "minimum_confidence": 0.7, "minimum_confidence_margin": 0.05}, + "runtime": {"allowed_algorithms": ["Hll", "Kll", "UnivMon"], "max_memory_bytes": null} + }); + let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture.clone()).unwrap(); + let plan = snapshot.compile().unwrap(); + eprintln!( + "UNIVMON_PLANNED {}", + serde_json::json!({"query_plan": plan.query_plan, "materializations": plan.precompute_plan.materializations, "lifecycle_estimates": plan.lifecycle_estimates, "executable_dags": plan.precompute_plan.executable_dags, "observation": observation}) + ); + assert!( + plan.precompute_plan + .materializations + .iter() + .any(|m| m.aggregation_type == asap_types::AggregationType::UnivMon), + "{plan:#?}" + ); + // Removing only entropy evidence must leave the L2 path executable. + let mut missing_entropy = fixture.clone(); + for row in missing_entropy["implementation"]["erp"]["artifact"]["records"] + .as_array_mut() + .unwrap() + { + row["error_metrics"] + .as_object_mut() + .unwrap() + .remove("max_frequency_entropy_absolute_bits_error"); + } + let missing = serde_json::from_value::(missing_entropy) + .unwrap() + .compile() + .unwrap(); + use control_plane::query_plan::{QueryPlanNode, QueryReadout}; + assert!(missing + .query_plan + .entries + .values() + .flat_map(|e| e.nodes.values()) + .any(|node| matches!( + node, + QueryPlanNode::SummaryEstimate { + query: QueryReadout::FrequencyL2, + .. + } + ))); + let entropy = missing + .query_plan + .entries + .values() + .find(|e| e.canonical_query.starts_with("entropy_over_time")) + .unwrap(); + assert!( + entropy.nodes.values().any(|node| matches!( + node, + QueryPlanNode::ExactFallback { .. } | QueryPlanNode::ExternalExact { .. } + )), + "{entropy:#?}" + ); + assert!(!entropy.nodes.values().any(|node| matches!( + node, + QueryPlanNode::SummaryEstimate { + query: QueryReadout::FrequencyEntropy, + .. + } + ))); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let fallback_url = format!("http://{}", listener.local_addr().unwrap()); + let fallback = tokio::spawn(async move { + axum::serve( + listener, + Router::new().route("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/-/healthy", get(|| async { "healthy" })), + ) + .await + .unwrap(); + }); + let output = tempfile::tempdir().unwrap(); + let path = output.path().join("planning.json"); + std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap(); + let port = unused_port(); + let mut child = ChildGuard( + Command::new(env!("CARGO_BIN_EXE_data_plane")) + .args(["--profile", "asapquery", "--planning-snapshot"]) + .arg(&path) + .args([ + "--prometheus-server", + &fallback_url, + "--forward-unsupported-queries", + "--http-port", + &port.to_string(), + "--output-dir", + ]) + .arg(output.path()) + .args([ + "--precompute-allowed-lateness-ms", + "0", + "--precompute-flush-interval-ms", + "25", + ]) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(), + ); + let client = reqwest::Client::new(); + let backend = format!("http://127.0.0.1:{port}"); + wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as i64; + let base = now - now.rem_euclid(5000) - 20_000; + let samples: Vec<_> = raw + .iter() + .enumerate() + .map(|(i, v)| (base + 1 + i as i64, *v)) + .collect(); + assert_eq!( + remote_write( + &client, + &backend, + &WriteRequest { + timeseries: vec![series("erp_frequency", &samples)] + } + ) + .await, + 204 + ); + assert_eq!( + remote_write( + &client, + &backend, + &WriteRequest { + timeseries: vec![series("erp_frequency", &[(base + 15001, 0.0)])] + } + ) + .await, + 204 + ); + drain_precompute(&client, &backend).await; + for (i, query) in queries.iter().enumerate() { + let result = wait_for_warm_instant( + &client, + &backend, + query, + (base + 5000) as f64 / 1000.0, + &output.path().join("query_engine.log"), + ) + .await; + let estimate = first_value(&result, "value").unwrap(); + let error = (estimate - exact[i]).abs() / if i == 2 { 1.0 } else { exact[i] }; + assert!( + error <= 0.2, + "{query}: {result}, truth={}, error={error}", + exact[i] + ); + eprintln!( + "UNIVMON_WARM {}", + serde_json::json!({"query": query, "result": result, "truth": exact[i], "measured_error": error, "units": if i == 2 { "absolute_bits" } else { "relative" }}) + ); + } + fallback.abort(); +} diff --git a/docs/developer_docs/univmon-erp-process-validation.md b/docs/developer_docs/univmon-erp-process-validation.md new file mode 100644 index 000000000..b3e5849ef --- /dev/null +++ b/docs/developer_docs/univmon-erp-process-validation.md @@ -0,0 +1,71 @@ +# UnivMon readout evidence through production execution + +The backend ERP adapter now selects evidence by the logical readout. Sharing a +UnivMon state does not give all of its readouts the same accuracy guarantee. + +| Readout | Artifact metric | Bound units | +| --- | --- | --- | +| Distinct | `max_cardinality_relative_error` | Relative distinct-count error | +| Frequency L2 | `max_frequency_l2_relative_error` | Relative error of `sqrt(sum frequency(value)^2)` | +| Frequency entropy | `max_frequency_entropy_absolute_bits_error` | Absolute Shannon entropy error, in bits | +| Count | No empirical metric needed | Exact total unit weight | + +Sizing maps `AggIntent` to this contract; accuracy validation maps `SketchQuery` +to the same contract. Matching artifact rows still pass implementation, runtime +parameters, trial count and bounded shape checks. Missing readout evidence leaves +that readout to theoretical sizing or exact fallback. It cannot borrow another +readout's measurement. ERP v1 does not calibrate failure probability, so an +explicit epsilon/delta requirement cannot use these observations as a confidence +bound. + +The existing `error_metric` field remains available to generic ERP adapter +callers. Production KLL and UnivMon readouts use their canonical typed metric +mapping. The recognized UnivMon implementation tag is +`asap-sketchlib-univmon-standard-v1`: standard unit-frequency updates over sample +value identities. Terminal promotion updates and numeric-value L2 have different +semantics and are not covered. + +## Reproduce the correctness fixture + +Run from the repository root: + +```sh +cargo test -p data_plane --test asapquery_compatibility_process_e2e measured_readout_evidence -- --nocapture +``` + +The test measures two configurations using the real runtime accumulator, ten +separate key populations, 128 distinct values and 1,338 unit updates per trial. +Each trial merges two panes. The artifact records maximum observed error and +serialized retained state size; CPU is not measured and its objective weight is +zero. These are local correctness measurements, not a throughput, latency, +resident-memory or general accuracy certification. + +| Configuration | Distinct relative error | L2 relative error | Entropy absolute bits error | +| --- | ---: | ---: | ---: | +| heap 8, columns 128, rows 5, layers 4 | 0.984375 | 0.0179301 | 0.545845 | +| heap 128, columns 1024, rows 5, layers 4 | 0.0078125 | 0.00003754 | 0.003554 | + +The observer fits the held-out frequency population and the normal Planner +chooses configurations. With a 0.2 bound in each readout's stated units: + +- Distinct selects HLL with precision 10 through its existing theoretical model. +- Frequency L2 selects the smaller UnivMon configuration. +- Entropy selects the larger UnivMon configuration. + +The resulting physical plan is installed in a real backend process. Remote +write feeds held-out keys, the production drain endpoint completes finite replay, +and all three queries execute through the warm ASAP path. The fallback service +only implements health checks, so it cannot supply query results. Held-out errors +were 2.054% for HLL distinct, 0.7304% for UnivMon L2, and approximately +3.55e-15 bits for entropy. This demonstrates selection and execution, not speedup. + +Removing only the entropy metric from the artifact makes entropy fall back while +preserving the L2 materialization. The test emits `UNIVMON_MEASURED`, +`UNIVMON_PLANNED`, and `UNIVMON_WARM` JSON records containing the artifact, observed +shape, selected parameters, installed DAG identities, lifecycle estimates and +actual query results. + +The three queries need different physical configurations in this fixture; they +do not demonstrate a single shared state. Planner may share compatible equal- +parameter states. The finite replay barrier also does not establish continuous +multi-worker population completeness, which is a separate publication contract.