fix: reject stored outputs without one consistent Planner producer

Materializations no longer carry their input filter, so an output's
population comes only from the DAG node that produces it. Plan
validation now rejects:

- an output produced by several DAGs that disagree on the update or on
  the population (metric and canonical filter) they read;
- a raw time-series output that has no Planner producer in a plan that
  carries DAGs, which would otherwise be read as unfiltered;
- a bound raw producer whose input is not a source scan.

Also keeps plain OTLP sketch envelopes out of shared (Hydra) states,
restores the single-catalog descriptor test, and drops comments that
still named removed fields.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: zzylol

crates/asap_summary_state/src/summary_kernels/factory.rs

@@ -615,9 +615,8 @@ pub enum TopkWeight {
 /// (`get_topk_keys` / `topk_heap_items`), which is what `topk(...)` queries
 /// need.
 ///
-/// The key is the configured group-by (`aggregated_labels`) value vector —
-/// e.g. `host` — formed by `extract_aggregated_key_from_series` in the worker,
-/// NOT the hardcoded metric label `item`. The accumulated quantity is selected
+/// The key is the Planner update's item value (e.g. the series identity for
+/// PromQL Top-K), not a hardcoded metric label `item`. The accumulated quantity is selected
 /// by [`TopkWeight`]:
 ///   * `Value` → `inner.update(key, value)` adds the datapoint value (Σ value).
 ///   * `Count` → `inner.update(key, 1.0)` adds one per event (Σ count).
@@ -862,7 +861,7 @@ fn cms_dims(params: &SketchParams) -> (usize, usize) {
             (*depth as usize, *width as usize)
         }
         other => unreachable!(
-            "accumulator_spec() paired SketchAlgorithm::Cms/CountSketch with unexpected params: {other:?}"
+            "SketchKind paired SketchAlgorithm::Cms/CountSketch with unexpected params: {other:?}"
         ),
     }
 }
@@ -882,7 +881,7 @@ fn cms_heap_dims(params: &SketchParams) -> (usize, usize, usize) {
             heap_size,
         } => (*depth as usize, *width as usize, *heap_size as usize),
         other => unreachable!(
-            "accumulator_spec() paired a WithHeap SketchAlgorithm with unexpected params: {other:?}"
+            "SketchKind paired a WithHeap SketchAlgorithm with unexpected params: {other:?}"
         ),
     }
 }

crates/asap_types/src/key_by_label_names.rs

@@ -3,7 +3,7 @@
 //! Formerly `promql_utilities::data_model::key_by_label_names` — moved
 //! here for the same reason as [`crate::Statistic`]: `asap_types`
 //! (`PrecomputeMaterialization::grouping_labels`, `PolicyFingerprint`,
-//! `PolicyRegistry`, `capability_matching`) is its real center of
+//! `capability_matching`) is its real center of
 //! gravity and the shared foundation both `control_plane`'s ecosystem
 //! and `data_plane` can depend on without a cycle. Closer to a runtime
 //! index key than a planning IR node — ASAPController's

crates/asap_types/src/precompute_plan.rs

@@ -314,6 +314,11 @@ pub enum PrecomputePlanError {
     InvalidSchema { schema_id: String },
     #[error("materialization {0} uses a summary family unsupported by the runtime schema")]
     UnsupportedFamily(u64),
+    #[error("materialization {materialization} has no consistent Planner computation: {reason}")]
+    InvalidComputation {
+        materialization: u64,
+        reason: String,
+    },
     #[error("materialization {materialization} has an invalid window layout: {reason}")]
     InvalidWindowLayout {
         materialization: u64,
@@ -466,7 +471,9 @@ impl PrecomputePlan {
 
     /// The Planner SummaryAgg node that produces `output`, with the source
     /// expression feeding it when that input is a raw source. `None` when no
-    /// installed DAG produces the output with a SummaryAgg.
+    /// installed DAG produces the output with a SummaryAgg. Every DAG that
+    /// produces the output must agree on both, or the plan is rejected: the
+    /// runtime routes, filters and catalogs the output once.
     #[allow(clippy::type_complexity)]
     pub fn summary_producer(
         &self,
@@ -479,6 +486,10 @@ impl PrecomputePlan {
         String,
     > {
         use planner_types::post_asap::PostAsapOperatorPayload;
+        let mut selected: Option<(
+            planner_types::post_asap::PostAsapDagNode,
+            Option<planner_types::pre_asap::QueryExpr>,
+        )> = None;
         for installed in self.executable_dags.values() {
             let Some(node) = installed.binding.nodes.iter().find_map(|(node, binding)| {
                 matches!(binding, crate::executable_plan::BackendNodeBinding::Materialization { stored_output } if *stored_output == output).then_some(*node)
@@ -508,9 +519,39 @@ impl PrecomputePlan {
                     }),
                 _ => None,
             };
-            return Ok(Some((producer.clone(), source)));
+            if let Some((first, first_source)) = &selected {
+                // Shared outputs may be read by scans that project different
+                // columns; what must agree is the update and its population.
+                let population =
+                    |node: &planner_types::post_asap::PostAsapDagNode,
+                     source: &Option<planner_types::pre_asap::QueryExpr>| {
+                        let PostAsapOperatorPayload::SummaryAgg { family, .. } = &node.payload
+                        else {
+                            return None;
+                        };
+                        source.as_ref().map(|expression| {
+                            raw_time_series_input_contract(
+                                expression,
+                                matches!(family, SummaryFamilyType::ExactAggregate(..)),
+                            )
+                            .map(|(metric, _, filter)| {
+                                (metric, crate::utils::normalize_spatial_filter(&filter))
+                            })
+                        })
+                    };
+                if first.payload != producer.payload
+                    || population(first, first_source) != population(producer, &source)
+                {
+                    return Err(format!(
+                        "stored output {} is produced by DAGs that disagree on its computation",
+                        output.as_u64()
+                    ));
+                }
+            } else {
+                selected = Some((producer.clone(), source));
+            }
         }
-        Ok(None)
+        Ok(selected)
     }
 
     /// Canonical population predicate of `config`'s input: the typed table
@@ -523,8 +564,16 @@ impl PrecomputePlan {
         if config.table_name.is_some() || config.derived_input.is_some() {
             return Ok(table);
         }
-        let Some((_, Some(expression))) = self.summary_producer(config.stored_output_id)? else {
-            return Ok(String::new());
+        let expression = match self.summary_producer(config.stored_output_id)? {
+            Some((_, Some(expression))) => expression,
+            // A bound raw output whose input cannot be read must not be
+            // treated as unfiltered.
+            Some((_, None)) => {
+                return Err("raw output's Planner producer does not read a source scan".into())
+            }
+            // Only plans without a Planner DAG for this output (imported or
+            // fixture outputs) have no predicate to read.
+            None => return Ok(String::new()),
         };
         let family = self.state_family(config.stored_output_id);
         let (_, _, filter) = raw_time_series_input_contract(
@@ -1033,6 +1082,28 @@ impl PrecomputePlan {
                     schema.materialization.as_u64(),
                 ));
             }
+            let invalid_computation = |reason: String| PrecomputePlanError::InvalidComputation {
+                materialization: schema.materialization.as_u64(),
+                reason,
+            };
+            let raw =
+                materialization.table_name.is_none() && materialization.derived_input.is_none();
+            // Once a plan carries Planner DAGs, a raw output's input predicate
+            // exists only on its producer; an unbound output would silently
+            // read every series of its metric.
+            if raw
+                && !self.executable_dags.is_empty()
+                && self
+                    .summary_producer(schema.materialization)
+                    .map_err(invalid_computation)?
+                    .is_none()
+            {
+                return Err(invalid_computation(
+                    "raw output has no Planner producer".into(),
+                ));
+            }
+            self.population_filter(materialization)
+                .map_err(invalid_computation)?;
             let source = materialization.table_name.as_ref().map_or_else(
                 || Source::TimeSeries {
                     metric: materialization.metric.clone(),

crates/asap_types/src/summary_catalog.rs

@@ -583,14 +583,22 @@ mod tests {
                 GroupingStrategy::PerSubpopulationInstance,
             )
         };
-        let a = with_family(1, 1, &[config("requests", "", 60)], &kll(100)).unwrap();
-        let b = with_family(1, 1, &[config("requests", "", 60)], &kll(200)).unwrap();
-        assert_ne!(a.summary_descriptors, b.summary_descriptors);
-        assert_eq!(a.data_descriptors, b.data_descriptors);
-        let catalog = a;
-        assert_eq!(catalog.summary_descriptors.len(), 1);
+        let (small, large) = (kll(100), kll(200));
+        let output = |family: &SummaryFamilyType| {
+            let (mut config, _) = config("requests", "", 60);
+            config.allocate_stored_output_id(family);
+            config
+        };
+        let (a, b) = (output(&small), output(&large));
+        let catalog = SummaryCatalog::from_outputs(
+            1,
+            1,
+            vec![(&a, &small, String::new()), (&b, &large, String::new())],
+        )
+        .unwrap();
+        assert_eq!(catalog.summary_descriptors.len(), 2);
         assert_eq!(catalog.data_descriptors.len(), 1);
-        assert_eq!(catalog.outputs.len(), 1);
+        assert_eq!(catalog.outputs.len(), 2);
     }
 
     // Construction order cannot affect the published snapshot bytes.

data_plane/src/drivers/ingest/otel.rs

@@ -1924,6 +1924,14 @@ fn derive_sketch_policy_fp(
         if bound_output.is_some_and(|output| output != config.stored_output_id)
             || config.metric != metric
             || &keys != group_by_keys
+            // A plain sketch envelope never fills a shared (Hydra) state.
+            || matches!(
+                snap.state_family(config.stored_output_id),
+                Some(planner_types::post_asap::SummaryFamilyType::Sketch(
+                    _,
+                    planner_types::post_asap::GroupingStrategy::SharedMultiSubpopulation { .. }
+                ))
+            )
             || snap
                 .agg_kind(config.stored_output_id)
                 .is_none_or(|installed| installed.canonical_string() != expected)

data_plane/src/storage_engines/types/installed_precompute_plan.rs

@@ -262,6 +262,77 @@ impl Default for InstalledPrecomputePlan {
 
 #[cfg(test)]
 mod tests {
+    fn filtered_plan(job: &str) -> control_plane::physical::compiler::PrecomputePlan {
+        let mut snapshot: serde_json::Value = serde_json::from_str(include_str!(
+            "../../../../docs/examples/asapquery-planning-snapshot.json"
+        ))
+        .unwrap();
+        snapshot["query_workload"]["repeating_queries"][0]["query"] =
+            serde_json::json!(format!("quantile_over_time(0.99, m{{job=\"{job}\"}}[1m])"));
+        crate::tests::test_utilities::planning::quoted_snapshot(
+            serde_json::from_value(snapshot).unwrap(),
+            false,
+        )
+        .compile_promql()
+        .unwrap()
+        .precompute_plan
+    }
+
+    /// Rebind every materialization node of `plan`'s DAGs to `output`.
+    fn bind_all(plan: &mut control_plane::physical::compiler::PrecomputePlan, output: u64) {
+        for installed in plan.executable_dags.values_mut() {
+            for binding in installed.binding.nodes.values_mut() {
+                if let asap_types::executable_plan::BackendNodeBinding::Materialization {
+                    stored_output,
+                } = binding
+                {
+                    *stored_output = asap_types::sds::StoredOutputId(output);
+                }
+            }
+        }
+    }
+
+    // Two DAGs that read different populations cannot share one stored output.
+    #[test]
+    fn producers_disagreeing_on_population_are_rejected() {
+        let mut plan = filtered_plan("a");
+        let output = plan.materializations[0].stored_output_id.as_u64();
+        let mut other = filtered_plan("b");
+        bind_all(&mut other, output);
+        let (_, mut dag) = other.executable_dags.into_iter().next().unwrap();
+        dag.document.query_id = "other-population".into();
+        plan.executable_dags.insert("other-population".into(), dag);
+        let error = plan.validate().unwrap_err().to_string();
+        assert!(error.contains("disagree on its computation"), "{error}");
+    }
+
+    // A raw output with no Planner producer is not read as unfiltered.
+    #[test]
+    fn unbound_raw_output_is_rejected() {
+        let mut plan = filtered_plan("a");
+        let mut unbound = plan.materializations[0].clone();
+        unbound.metric = "unbound".into();
+        unbound.stored_output_id = asap_types::sds::StoredOutputId(1);
+        let mut schema = plan.schemas[0].clone();
+        schema.materialization = unbound.stored_output_id;
+        schema.schema_id = schema.schema_id.replace(
+            &plan.materializations[0]
+                .stored_output_id
+                .as_u64()
+                .to_string(),
+            "1",
+        );
+        schema.stored_output_reference =
+            asap_types::sds::StoredOutputReference::for_output(unbound.stored_output_id);
+        plan.materializations.push(unbound);
+        plan.schemas.push(schema);
+        let error = plan.validate().unwrap_err().to_string();
+        assert!(
+            error.contains("raw output has no Planner producer"),
+            "{error}"
+        );
+    }
+
     // A raw output's input predicate, pane convention and stored state kind
     // come from its installed Planner producer, not from the materialization.
     #[test]

data_plane/src/storage_engines/types/precomputed_output.rs

@@ -78,7 +78,7 @@ impl PrecomputedOutput {
     ///
     /// `policy_fp` is the content-addressed handle on the source
     /// [`asap_types::PrecomputeMaterialization`]; sinks use it to look up the
-    /// config via `PolicyRegistry::get(policy_fp)`. Construction sites
+    /// installed output by it. Construction sites
     /// that lack a source config (raw-mode fast-path) pass
     /// [`PolicyFingerprint::UNSET`]; sinks then skip the output.
     pub fn new(

data_plane/src/tests/accumulator_fixture.rs

@@ -41,20 +41,15 @@ fn topk_weight(update: &SummaryUpdate) -> TopkWeight {
 // Factory function
 // ---------------------------------------------------------------------------
 
-/// Read the KLL `k` out of `SketchParams::Kll`. `accumulator_spec()`
-/// always builds a `SketchKind` whose `SketchAlgorithm::Kll` is paired with
-/// `SketchParams::Kll`, so the
-/// other arm is unreachable from a `spec` this module builds itself.
+/// Read the KLL `k` out of `SketchParams::Kll`; a valid `SketchKind`
+/// pairs `SketchAlgorithm::Kll` with no other params.
 #[cfg(test)]
 fn kll_k(params: &SketchParams) -> u16 {
     match params {
-        // Lossless: `accumulator_spec()` only ever stores a value that
-        // already fit in `u16` (via `kll_k_param`'s own `u16::try_from`
-        // fallback) widened to `u32`.
         SketchParams::Kll { k } => *k as u16,
-        other => unreachable!(
-            "accumulator_spec() paired SketchAlgorithm::Kll with non-Kll params: {other:?}"
-        ),
+        other => {
+            unreachable!("SketchKind paired SketchAlgorithm::Kll with non-Kll params: {other:?}")
+        }
     }
 }
 
@@ -66,7 +61,7 @@ fn cms_dims(params: &SketchParams) -> (usize, usize) {
             (*depth as usize, *width as usize)
         }
         other => unreachable!(
-            "accumulator_spec() paired SketchAlgorithm::Cms/CountSketch with unexpected params: {other:?}"
+            "SketchKind paired SketchAlgorithm::Cms/CountSketch with unexpected params: {other:?}"
         ),
     }
 }
@@ -86,7 +81,7 @@ fn cms_heap_dims(params: &SketchParams) -> (usize, usize, usize) {
             heap_size,
         } => (*depth as usize, *width as usize, *heap_size as usize),
         other => unreachable!(
-            "accumulator_spec() paired a WithHeap SketchAlgorithm with unexpected params: {other:?}"
+            "SketchKind paired a WithHeap SketchAlgorithm with unexpected params: {other:?}"
         ),
     }
 }
@@ -97,7 +92,7 @@ fn ddsketch_alpha(params: &SketchParams) -> f64 {
     match params {
         SketchParams::DDSketch { alpha } => *alpha,
         other => unreachable!(
-            "accumulator_spec() paired SketchAlgorithm::DDSketch with non-DDSketch params: {other:?}"
+            "SketchKind paired SketchAlgorithm::DDSketch with non-DDSketch params: {other:?}"
         ),
     }
 }
@@ -187,9 +182,8 @@ pub fn create_fixture_accumulator(
         // Heap-bearing top-k variant (raw-input ingest path): route to the
         // real `CmsHeapAccumulatorUpdater` so the per-policy top-k heap is
         // BUILT (heap-less CMS could not answer `topk(...)` — recall 0).
-        // Keyed by the configured group-by `aggregated_labels` (e.g. `host`),
-        // ranked by Σ value per key by default (`weight_mode: value`), or Σ
-        // count for genuine frequency-top-k (`weight_mode: count`). The OTLP
+        // Keyed by the update's item, ranked by Σ value per key, or Σ count
+        // when the update weight is the constant 1. The OTLP
         // modified-sketch path builds the heap agent-side and uses
         // `SketchEnvelope` ingest, not this raw arm.
         (SummaryFamilyType::Sketch(kind, _), _)