feat(maintenance): activate complete canonical populations

.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

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

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::<Result<Vec<_>, _>>()?;
+            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()

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() {

data_plane/src/precompute_engine/maintenance_runtime.rs

@@ -286,6 +286,7 @@ impl PrecomputeOperatorRegistry<MaintenanceValue> 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::<String>().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();

data_plane/src/query_engines/asap_query_engine/summary_executor.rs

@@ -1045,7 +1045,18 @@ fn sketch_query_value(rs: &SummaryState, query: &SketchQuery) -> Result<f64, Sum
         SketchQuery::FrequencyL2 | SketchQuery::FrequencyEntropy => 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();