diff --git a/.github/workflows/mvp-ci.yml b/.github/workflows/mvp-ci.yml index fdd9896e..a2b9d4cb 100644 --- a/.github/workflows/mvp-ci.yml +++ b/.github/workflows/mvp-ci.yml @@ -34,8 +34,8 @@ jobs: uses: actions/checkout@v4 with: repository: ProjectASAP/asap_sketchlib - # PR #139: standard-update compatibility check used by UnivMon restore. - ref: c0de315754f9a6c77dd7a25aca0b2b62f0aec276 + # PR #140: explicit interpolated DDS readout; includes #139 UnivMon compatibility. + ref: 8c03d7c68b7150710c79e70a6421971275614561 path: asap_sketchlib - name: Install Rust components diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index c4967cc9..91d761e2 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1095,6 +1095,24 @@ impl PhysicalCompiler { }) } } + // Choose the population protocol before any source fingerprint or + // derived frontier binding is created. Only the actual selected + // global maintenance program and its raw inputs opt into it. + let mut canonical_nodes = std::collections::HashSet::new(); + if environment.target == PhysicalDeploymentTarget::BackendLocalRemoteWrite { + for state in &selected { + if matches!(&state.node.expr, SummaryExpr::SummaryAgg { + reduction: planner_types::pre_asap::Reduction::Reduce(keys), .. + } if keys.is_empty()) + { + if let Some(sources) = immutable_materialization_sources(&state.node) { + canonical_nodes.insert(Rc::as_ptr(&state.node) as usize); + canonical_nodes + .extend(sources.iter().map(|source| Rc::as_ptr(source) as usize)); + } + } + } + } for (ordinal, selected) in selected.into_iter().enumerate() { let mut branch_query = query.clone(); branch_query.window_secs = selected.window_secs.unwrap_or(query.window_secs); @@ -1179,6 +1197,10 @@ impl PhysicalCompiler { aggregation.window_secs = window_implementation.window_secs; let mut runtime_materialization = scoped_materialization(&aggregation, &selected.node)?; + if canonical_nodes.contains(&(Rc::as_ptr(&selected.node) as usize)) { + runtime_materialization.population_key_encoding = + asap_types::PopulationKeyEncoding::CanonicalLabelsV1; + } runtime_materialization.window_size = window_implementation.window_secs; runtime_materialization.slide_interval = window_implementation.slide_secs; runtime_materialization.window_type = @@ -3793,6 +3815,23 @@ mod tests { deployment.collector_ids.clear(); let plan = PhysicalCompiler.compile(workload, deployment).unwrap(); assert_eq!(plan.precompute_plan.materializations.len(), 3); + assert!(plan + .precompute_plan + .materializations + .iter() + .all(|config| config.population_key_encoding + == asap_types::PopulationKeyEncoding::CanonicalLabelsV1)); + let mut mixed = plan.precompute_plan.clone(); + mixed + .materializations + .iter_mut() + .find(|config| config.derived_input.is_none()) + .unwrap() + .population_key_encoding = asap_types::PopulationKeyEncoding::LegacyDelimited; + assert!( + mixed.validate().is_err(), + "canonical target cannot use legacy source identity" + ); let derived = plan .precompute_plan .materializations diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index 994b5a2b..4a6ce9d7 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -476,8 +476,21 @@ impl PrecomputePlan { if !valid_ingest { return Err(PrecomputePlanError::UnsupportedIngestEndpoint); } + let canonical_cohort_members: BTreeSet<_> = self + .materializations + .iter() + .filter(|config| !config.population_key_encoding.is_legacy()) + .filter_map(|config| config.derived_input.as_ref().map(|input| (config, input))) + .flat_map(|(config, input)| { + std::iter::once(config.policy_fingerprint().into()) + .chain(input.inputs.iter().copied()) + }) + .collect(); for config in &self.materializations { - if !config.population_key_encoding.is_legacy() { + if !config.population_key_encoding.is_legacy() + && (self.ingest.protocol != IngestProtocol::PrometheusRemoteWriteV1 + || !canonical_cohort_members.contains(&config.policy_fingerprint().into())) + { return Err(PrecomputePlanError::CatalogContract( "population key encoding is not supported by the installed runtime".into(), )); @@ -506,6 +519,21 @@ impl PrecomputePlan { .ok_or_else(invalid) }) .collect::, _>>()?; + if !config.population_key_encoding.is_legacy() { + if config.partitioning != Some(crate::sds::PopulationPartitioning::Grouped) + || !config.grouping_labels.is_empty() + || sources.iter().any(|source| { + source.population_key_encoding != config.population_key_encoding + }) + { + return Err(invalid()); + } + } else if sources + .iter() + .any(|source| !source.population_key_encoding.is_legacy()) + { + return Err(invalid()); + } validated_source_window_cohort(config, &sources)?; if sources .iter() diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs index f7e9ca50..81ae8fb9 100644 --- a/data_plane/src/precompute_engine/accumulator_factory.rs +++ b/data_plane/src/precompute_engine/accumulator_factory.rs @@ -58,6 +58,16 @@ macro_rules! impl_clone_accumulator_methods { /// This provides a uniform interface over all accumulator types so that the /// worker loop doesn't need to know which concrete type it's dealing with. pub trait AccumulatorUpdater: Send { + /// Validate an immutable maintenance input before an updater can silently + /// discard a value outside its representable domain. + fn validate_single_input(&self, value: f64) -> Result<(), String> { + if value.is_finite() { + Ok(()) + } else { + Err("accumulator input must be finite".into()) + } + } + /// Feed a single (value, timestamp_ms) pair — for SingleSubpopulation types. fn update_single(&mut self, value: f64, timestamp_ms: i64); @@ -332,6 +342,16 @@ impl DDSketchAccumulatorUpdater { } impl AccumulatorUpdater for DDSketchAccumulatorUpdater { + fn validate_single_input(&self, value: f64) -> Result<(), String> { + let (minimum, maximum) = + asap_sketchlib::sketches::ddsketch::ddsketch_indexable_bounds(self.alpha); + if value.is_finite() && value > 0.0 && value >= minimum && value <= maximum { + Ok(()) + } else { + Err("DDS maintenance input is outside its positive representable domain".into()) + } + } + fn update_single(&mut self, value: f64, _timestamp_ms: i64) { // sketch-core's DdSketch (the inner of DDSketchAccumulator) // exposes `update(f64)` for single-value ingestion. The @@ -1252,6 +1272,17 @@ mod tests { use asap_types::enums::WindowKind; use asap_types::AggregationType; + #[test] + fn immutable_dds_inputs_reject_nonpositive_and_unrepresentable_values() { + let updater = DDSketchAccumulatorUpdater::new(0.01); + for value in [-20.0, -0.0, 0.0, f64::NAN, f64::INFINITY, f64::MAX] { + assert!(updater.validate_single_input(value).is_err()); + } + for value in [0.5, 20.0, 40.0] { + assert!(updater.validate_single_input(value).is_ok()); + } + } + /// Both cardinality implementations consume values, with a single signed-zero identity. #[test] fn hll_and_univmon_raw_updates_share_value_identity() { diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 8fa31430..47fb0428 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -286,6 +286,7 @@ impl PrecomputeOperatorRegistry for OperatorAdapter<'_> { } for (timestamp_ms, value) in values.values().flatten() { let weight = evaluate_weight(&input.weight, *value, name)?; + updater.validate_single_input(weight)?; updater.update_single(weight, *timestamp_ms); } let timestamp = values @@ -3577,6 +3578,68 @@ mod tests { Err(error) if error.contains("one explicitly reduced output population"))); } + #[test] + fn dds_maintenance_rejects_nonpositive_population_before_returning_summary() { + use planner_types::post_asap::{GroupingStrategy, SummaryUpdate}; + use planner_types::pre_asap::{ColumnRef, Reduction}; + let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + let mut config = snapshot.compile().unwrap().precompute_plan.materializations[0].clone(); + config.aggregation_type = asap_types::AggregationType::DDSketch; + config.parameters.clear(); + config + .parameters + .insert("relative_accuracy".into(), "0.01".into()); + config.aggregation_sub_type.clear(); + config.grouping_labels = std::iter::empty::().collect(); + config.partitioning = Some(asap_types::sds::PopulationPartitioning::Grouped); + let family = config.accumulator_spec().unwrap().family; + let binding = BackendExecutableBinding { + nodes: BTreeMap::from([( + PostAsapNodeId(1), + BackendNodeBinding::Materialization { + summary_definition: config.policy_fingerprint().into(), + }, + )]), + query_sink: PostAsapNodeId(1), + query_plan_sink: asap_types::query_plan::QueryNodeId(1), + precompute_sinks: vec![PostAsapNodeId(1)], + }; + let configs = [config]; + let adapter = OperatorAdapter { + binding: &binding, + inputs: MaintenanceInputs::Frozen(&[]), + configs: &configs, + }; + let mut aggregate = node(1); + aggregate.payload = ExecutableOperatorPayload::SummaryAgg { + family, + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::by(vec![]), + grouping: GroupingStrategy::default(), + }; + for rejected in [-20.0, 0.0, f64::MAX] { + let rows = MaintenanceValue::Rows { + values: [("a", 20.0), ("b", rejected)] + .into_iter() + .map(|(group, value)| { + ( + BTreeMap::from([("instance".into(), group.into())]), + vec![(1000, value)], + ) + }) + .collect(), + name: "value".into(), + timestamped: true, + }; + assert!(matches!(adapter.execute(&aggregate, &[Arc::new(rows)]), + Err(error) if error.contains("positive representable domain"))); + } + } + #[test] fn admitted_slow_worker_can_publish_behind_another_workers_replay_floor() { let commits = CommitRegistry::default(); diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs index ab3389e0..0b3583e4 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs @@ -1045,7 +1045,18 @@ fn sketch_query_value(rs: &SummaryState, query: &SketchQuery) -> Result Err( SummaryExecutorError::Unsupported("frequency moment readout requires UnivMon"), ), - SketchQuery::Quantile { q } => Ok(rs.quantile(*q)), + SketchQuery::Quantile { q } => match rs { + // Typed PromQL/continuous-percentile readout uses interpolation; + // portable DDS `quantile` deliberately retains lower-rank parity. + SummaryState::Dd(sketch) => { + sketch + .quantile_interpolated(*q) + .ok_or(SummaryExecutorError::Unsupported( + "DDS interpolated quantile is unavailable", + )) + } + _ => Ok(rs.quantile(*q)), + }, SketchQuery::Cardinality => Ok(rs.cardinality()), // `key: ColumnRef::SampleValue, value: None` means "no specific // item" -- the bare bucket total. `key: Named(_), value: Some(v)` @@ -1949,6 +1960,40 @@ mod tests { assert!(result.is_empty()); } + #[test] + fn typed_dds_quantile_interpolates_without_changing_portable_rank_semantics() { + let mut sketch = asap_sketchlib::DdSketch::new(0.01); + assert!(sketch_query_value( + &SummaryState::Dd(sketch.clone()), + &SketchQuery::Quantile { q: 0.9 } + ) + .is_err()); + sketch.update(20.0); + for q in [0.0, 0.5, 0.9, 1.0] { + let value = sketch_query_value( + &SummaryState::Dd(sketch.clone()), + &SketchQuery::Quantile { q }, + ) + .unwrap(); + assert!((value - 20.0).abs() <= 0.2); + } + sketch.update(40.0); + assert!(sketch.quantile(0.9).unwrap() < 21.0); + for (q, expected) in [(0.0, 20.0), (0.5, 30.0), (0.9, 38.0), (1.0, 40.0)] { + let value = sketch_query_value( + &SummaryState::Dd(sketch.clone()), + &SketchQuery::Quantile { q }, + ) + .unwrap(); + assert!((value - expected).abs() <= expected * 0.01); + } + assert!(sketch_query_value( + &SummaryState::Dd(sketch), + &SketchQuery::Quantile { q: f64::NAN } + ) + .is_err()); + } + #[test] fn single_kll_sid_quantile_readout() { let idx = SketchStore::new(); diff --git a/data_plane/tests/support/immutable_maintenance_process.rs b/data_plane/tests/support/immutable_maintenance_process.rs index e329e9a2..eb191c0a 100644 --- a/data_plane/tests/support/immutable_maintenance_process.rs +++ b/data_plane/tests/support/immutable_maintenance_process.rs @@ -3,21 +3,26 @@ use super::*; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn single_source_maintenance_is_automatic_and_durable() { - run_maintenance_process(false).await; + run_maintenance_process(false, false).await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn two_source_maintenance_is_automatic_and_durable() { - run_maintenance_process(true).await; + run_maintenance_process(true, false).await; } -async fn run_maintenance_process(multi_source: bool) { +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn complete_group_maintenance_is_automatic_and_durable() { + run_maintenance_process(true, true).await; +} + +async fn run_maintenance_process(multi_source: bool, distinct_groups: bool) { let query = if multi_source { "quantile(0.9, sum_over_time(immutable_value[1m]) + sum_over_time(immutable_other[1m]))" } else { "quantile(0.9, sum_over_time(immutable_value[1m]))" }; - let expected = if multi_source { 20.0 } else { 10.0 }; + let base_expected = if multi_source { 20.0 } else { 10.0 }; let mut fixture: Value = serde_json::from_str(include_str!( "../../../docs/examples/asapquery-compatibility-demo-snapshot.json" )) @@ -83,12 +88,18 @@ async fn run_maintenance_process(multi_source: bool) { storage_routing: None, adaptation_evidence: vec![], }; - // Two independent deployments: singleton is supported; a second physical - // input series must never be mistaken for a complete singleton population. - for (count, missing_source) in [(1, false), (2, false), (1, true)] { + // Complete canonical populations are supported; missing source/group sets + // must fail closed before publishing any global output. + for (count, missing_source) in [(1, true), (2, true), (1, false), (2, false)] { + let expected = if distinct_groups && count == 2 { + 38.0 + } else { + base_expected + }; if missing_source && !multi_source { continue; } + eprintln!("IMMUTABLE_CASE count={count} missing_source_or_group={missing_source}"); let mut directory = tempfile::tempdir().unwrap(); eprintln!("IMMUTABLE_PROCESS_ARTIFACT {}", directory.path().display()); directory.disable_cleanup(true); @@ -146,19 +157,27 @@ async fn run_maintenance_process(multi_source: bool) { ("instance", if i == 0 { "a" } else { "b" }), ("job", "worker"), ], - &[(1_000, 2.0), (2_000, 3.0), (60_000, 5.0)], + &[ + (1_000, if distinct_groups && i == 1 { 4.0 } else { 2.0 }), + (2_000, if distinct_groups && i == 1 { 6.0 } else { 3.0 }), + (60_000, if distinct_groups && i == 1 { 10.0 } else { 5.0 }), + ], ) }) .collect(); - if multi_source && !missing_source { - for i in 0..count { + if multi_source && (!missing_source || count == 2) { + for i in 0..if missing_source { 1 } else { count } { series.push(series_with_labels( "immutable_other", &[ ("instance", if i == 0 { "a" } else { "b" }), ("job", "worker"), ], - &[(1_000, 2.0), (2_000, 3.0), (60_000, 5.0)], + &[ + (1_000, if distinct_groups && i == 1 { 4.0 } else { 2.0 }), + (2_000, if distinct_groups && i == 1 { 6.0 } else { 3.0 }), + (60_000, if distinct_groups && i == 1 { 10.0 } else { 5.0 }), + ], )); } } @@ -171,7 +190,7 @@ async fn run_maintenance_process(multi_source: bool) { .send() .await .unwrap(); - if count == 1 && !missing_source { + if !missing_source { assert!( drain.status().is_success(), "{}", @@ -187,10 +206,10 @@ async fn run_maintenance_process(multi_source: bool) { .json() .await .unwrap(); - if count == 2 || missing_source { + if missing_source { assert!( !is_warm(&response), - "multi-series population was incorrectly admitted: {response}" + "incomplete source group set was incorrectly admitted: {response}" ); continue; } @@ -211,7 +230,31 @@ async fn run_maintenance_process(multi_source: bool) { .unwrap(); assert!( estimate.is_finite() && (estimate - expected).abs() / expected <= max_relative_error, - "selected singleton quantile exceeded its value contract: {response}" + "selected population quantile exceeded its value contract: {response}" + ); + let part_ids = || { + let mut ids = std::fs::read_dir(disk.join("sketch_index/parts")) + .unwrap() + .map(|entry| entry.unwrap().file_name()) + .collect::>(); + ids.sort(); + ids + }; + let before_retry_parts = part_ids(); + let repeated = client + .post(format!("{backend}/api/v1/precompute/drain")) + .send() + .await + .unwrap(); + assert!( + repeated.status().is_success(), + "{}", + repeated.text().await.unwrap() + ); + assert_eq!( + part_ids(), + before_retry_parts, + "repeated completion published extra parts" ); drop(first); let port = unused_port(); @@ -234,6 +277,11 @@ async fn run_maintenance_process(multi_source: bool) { .unwrap(); assert!(is_warm(&after), "{after}"); assert_eq!(after["data"]["result"], response["data"]["result"]); + assert_eq!( + part_ids(), + before_retry_parts, + "restart published extra immutable parts" + ); assert_ne!( remote_write( &client,