From 70e3392d8eed78cf1d10efa54d27ae74dfbb3390 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 17:47:24 -0600
Subject: [PATCH 1/8] feat(sds): identify entity-preserving population
 partitions

---
 crates/asap_types/src/aggregation_config.rs   | 14 +++++
 crates/asap_types/src/policy_fingerprint.rs   |  4 ++
 .../asap_types/src/precompute_plan/catalog.rs |  3 +-
 crates/asap_types/src/sds.rs                  | 56 +++++++++++++++++++
 crates/asap_types/src/summary_catalog.rs      |  3 +-
 5 files changed, 78 insertions(+), 2 deletions(-)

diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs
index 9f16833df..d94dbf145 100644
--- a/crates/asap_types/src/aggregation_config.rs
+++ b/crates/asap_types/src/aggregation_config.rs
@@ -100,6 +100,8 @@ pub struct PrecomputeMaterialization {
     pub aggregation_sub_type: String,
     pub parameters: HashMap<String, Value>,
     pub grouping_labels: KeyByLabelNames,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub partitioning: Option<crate::sds::PopulationPartitioning>,
     pub aggregated_labels: KeyByLabelNames,
     pub rollup_labels: KeyByLabelNames,
     pub original_yaml: String,
@@ -189,6 +191,7 @@ impl PrecomputeMaterialization {
             aggregation_sub_type,
             parameters,
             grouping_labels,
+            partitioning: None,
             aggregated_labels,
             rollup_labels,
             original_yaml,
@@ -317,6 +320,11 @@ impl PrecomputeMaterialization {
             table_name,
             value_column,
         );
+        config.partitioning = data
+            .get("partitioning")
+            .filter(|value| !value.is_null())
+            .map(|value| serde_json::from_value(value.clone()))
+            .transpose()?;
         config.pane_origin_ms = pane_origin_ms;
         Ok(config)
     }
@@ -456,6 +464,11 @@ impl PrecomputeMaterialization {
             table_name,
             value_column,
         );
+        config.partitioning = aggregation_data
+            .get("partitioning")
+            .filter(|value| !value.is_null())
+            .map(|value| serde_yaml::from_value(value.clone()))
+            .transpose()?;
         config.pane_origin_ms = pane_origin_ms;
         Ok(config)
     }
@@ -469,6 +482,7 @@ impl SerializableToSink for PrecomputeMaterialization {
             "aggregationType": self.aggregation_type,
             "aggregationSubType": self.aggregation_sub_type,
             "parameters": self.parameters,
+            "partitioning": self.partitioning,
             "originalYaml": self.original_yaml,
             "windowSize": self.window_size,
             "slideInterval": self.slide_interval,
diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs
index eea167c87..3ab2f0d00 100644
--- a/crates/asap_types/src/policy_fingerprint.rs
+++ b/crates/asap_types/src/policy_fingerprint.rs
@@ -114,6 +114,10 @@ impl PolicyFingerprint {
         }
         buf.push(0);
 
+        if let Some(partitioning) = cfg.partitioning {
+            buf.extend_from_slice(format!("partition:{partitioning:?}\0").as_bytes());
+        }
+
         // 5. grouping_labels (already sorted at construction per
         //    KeyByLabelNames invariant; encode as `,`-joined list)
         for l in &cfg.grouping_labels.labels {
diff --git a/crates/asap_types/src/precompute_plan/catalog.rs b/crates/asap_types/src/precompute_plan/catalog.rs
index 48fd1c411..76bc1ab97 100644
--- a/crates/asap_types/src/precompute_plan/catalog.rs
+++ b/crates/asap_types/src/precompute_plan/catalog.rs
@@ -86,7 +86,8 @@ impl PrecomputePlan {
                 .map_or(ValueProjectionIdentity::SampleValue, |name| {
                     ValueProjectionIdentity::Column { name: name.clone() }
                 });
-            if data.source != expected_source
+            if data.partitioning != config.partitioning
+                || data.source != expected_source
                 || data.value_projection != expected_projection
                 || data.population_filter_canonical
                     != crate::utils::normalize_spatial_filter(&config.spatial_filter)
diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs
index b54c50299..a85422bc9 100644
--- a/crates/asap_types/src/sds.rs
+++ b/crates/asap_types/src/sds.rs
@@ -698,6 +698,15 @@ pub enum ValueProjectionIdentity {
     Column { name: String },
 }
 
+/// Whether a materialization preserves source entities or pools a population.
+/// Grouped label names remain in `DataDescriptor::group_by_keys`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum PopulationPartitioning {
+    PerEntity,
+    Grouped,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct DataDescriptor {
@@ -706,6 +715,8 @@ pub struct DataDescriptor {
     pub value_projection: ValueProjectionIdentity,
     pub population_filter_canonical: String,
     pub group_by_keys: BTreeSet<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub partitioning: Option<PopulationPartitioning>,
     /// Versioned contract for timestamp interpretation and
     /// missing/duplicate/invalid observation handling.
     pub observation_semantics: String,
@@ -765,6 +776,7 @@ impl DataDescriptor {
             &population_filter_canonical,
             &group_by_keys,
             &observation_semantics,
+            None,
         );
         Self {
             id,
@@ -772,9 +784,22 @@ impl DataDescriptor {
             value_projection,
             population_filter_canonical,
             group_by_keys,
+            partitioning: None,
             observation_semantics,
         }
     }
+    pub fn with_partitioning(mut self, partitioning: Option<PopulationPartitioning>) -> Self {
+        self.partitioning = partitioning;
+        self.id = data_descriptor_id(
+            &self.source,
+            &self.value_projection,
+            &self.population_filter_canonical,
+            &self.group_by_keys,
+            &self.observation_semantics,
+            partitioning,
+        );
+        self
+    }
     pub fn id(&self) -> &DataDescriptorId {
         &self.id
     }
@@ -786,6 +811,7 @@ impl DataDescriptor {
                 &self.population_filter_canonical,
                 &self.group_by_keys,
                 &self.observation_semantics,
+                self.partitioning,
             )
         {
             return Err(SdsError("data descriptor ID/content mismatch".into()));
@@ -799,6 +825,7 @@ fn data_descriptor_id(
     filter: &str,
     group_by: &BTreeSet<String>,
     observation_semantics: &str,
+    partitioning: Option<PopulationPartitioning>,
 ) -> DataDescriptorId {
     // Length framing keeps distinct typed sources, projections, predicates,
     // and grouping keys collision-free in the content identity.
@@ -811,6 +838,9 @@ fn data_descriptor_id(
         projection.len(),
         filter.len()
     );
+    if let Some(partitioning) = partitioning {
+        key.push_str(&format!("|partition:{partitioning:?}"));
+    }
     for name in group_by {
         key.push_str(&format!("|{}:{name}", name.len()));
     }
@@ -1286,3 +1316,29 @@ mod tests {
         );
     }
 }
+
+#[cfg(test)]
+mod partition_identity_tests {
+    use super::*;
+    #[test]
+    fn entity_and_global_population_have_distinct_identity() {
+        let legacy = DataDescriptor::new_typed(
+            DataSourceIdentity::TimeSeries { metric: "m".into() },
+            ValueProjectionIdentity::SampleValue,
+            "",
+            Vec::<String>::new(),
+            "v1",
+        );
+        let entity = legacy
+            .clone()
+            .with_partitioning(Some(PopulationPartitioning::PerEntity));
+        let grouped = legacy
+            .clone()
+            .with_partitioning(Some(PopulationPartitioning::Grouped));
+        assert_ne!(entity.id, grouped.id);
+        assert_ne!(entity.id, legacy.id);
+        assert_eq!(legacy.clone().with_partitioning(None).id, legacy.id);
+        entity.validate().unwrap();
+        grouped.validate().unwrap();
+    }
+}
diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs
index 90095a330..7ea42cca0 100644
--- a/crates/asap_types/src/summary_catalog.rs
+++ b/crates/asap_types/src/summary_catalog.rs
@@ -123,7 +123,8 @@ impl SummaryCatalog {
                     crate::utils::normalize_spatial_filter(&config.spatial_filter),
                     config.grouping_labels.labels.clone(),
                     "asap.timestamped-observations.v2",
-                );
+                )
+                .with_partitioning(config.partitioning);
                 Ok((
                     config.policy_fingerprint(),
                     summary,

From b836fea89671748146329c01372136d9f4abd873 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 17:56:24 -0600
Subject: [PATCH 2/8] feat: preserve raw entity partitions selected by the DAG

---
 control_plane/src/physical/compiler.rs        | 110 +++++++++++-------
 crates/asap_types/src/precompute_plan.rs      |  43 +++++++
 .../drivers/ingest/prometheus_remote_write.rs |  17 +--
 3 files changed, 119 insertions(+), 51 deletions(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 83b60d836..cf459fd41 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -1513,9 +1513,12 @@ fn has_unsafe_raw_entity_leaf(
                         _
                     )
                 );
+                let scalar_series_input = matches!(&node.expr,
+                    SummaryExpr::SummaryAgg { input, .. }
+                    if input.item.is_none() && matches!(&input.weight, planner_types::post_asap::SummaryInputExpr::Column(planner_types::pre_asap::ColumnRef::SampleValue)));
                 return matches!(reduction, Reduction::PerEntity)
                     && !pooling
-                    && !preserves_series_state;
+                    && !(preserves_series_state || scalar_series_input);
             }
             let additive_reduction = matches!(reduction, Reduction::Reduce(_))
                 && matches!(family, SummaryFamilyType::ExactAggregate(ExactKind::Sum, _))
@@ -1872,7 +1875,7 @@ impl PhysicalCompiler {
                     environment.target,
                 );
                 let precompute_materialization =
-                    aggregation_config_for_materialization(&aggregation)?;
+                    scoped_materialization(&aggregation, &selected.node)?;
                 let materialization = precompute_materialization.policy_fingerprint();
                 let state_consumers = consumers[&materialization]
                     .iter()
@@ -1904,7 +1907,7 @@ impl PhysicalCompiler {
                 // from the selected storage representation.
                 aggregation.window_secs = window_implementation.window_secs;
                 let mut runtime_materialization =
-                    aggregation_config_for_materialization(&aggregation)?;
+                    scoped_materialization(&aggregation, &selected.node)?;
                 runtime_materialization.window_size = window_implementation.window_secs;
                 runtime_materialization.slide_interval = window_implementation.slide_secs;
                 runtime_materialization.window_type =
@@ -2859,10 +2862,12 @@ fn retained_partition_count(
     // Reset-aware and min/max state remains source-series scoped even with an
     // empty output grouping. Grouped states have at most one partition per
     // input series. Other empty groupings are the Reduce([]) global singleton.
-    if matches!(
-        materialization.aggregation_type,
-        A::Increase | A::MultipleIncrease | A::MinMax | A::MultipleMinMax
-    ) || !materialization.grouping_labels.labels.is_empty()
+    if materialization.partitioning == Some(asap_types::sds::PopulationPartitioning::PerEntity)
+        || matches!(
+            materialization.aggregation_type,
+            A::Increase | A::MultipleIncrease | A::MinMax | A::MultipleMinMax
+        )
+        || !materialization.grouping_labels.labels.is_empty()
     {
         u128::from(input_cardinality.unwrap_or(1).max(1))
     } else {
@@ -3213,6 +3218,25 @@ fn physical_aggregation(
 /// content-addressed materialization contract. This is the one conversion
 /// shared by the physical compiler and the compatibility replanner; it does
 /// not create a second registry or wire plan.
+fn scoped_materialization(
+    aggregation: &BackendAggregation,
+    node: &SummaryNode,
+) -> anyhow::Result<asap_types::PrecomputeMaterialization> {
+    let mut config = aggregation_config_for_materialization(aggregation)?;
+    let SummaryExpr::SummaryAgg { reduction, .. } = &node.expr else {
+        anyhow::bail!("materialization lacks SummaryAgg partition contract");
+    };
+    config.partitioning = Some(match reduction {
+        planner_types::pre_asap::Reduction::PerEntity => {
+            asap_types::sds::PopulationPartitioning::PerEntity
+        }
+        planner_types::pre_asap::Reduction::Reduce(_) => {
+            asap_types::sds::PopulationPartitioning::Grouped
+        }
+    });
+    Ok(config)
+}
+
 pub(crate) fn aggregation_config_for_materialization(
     aggregation: &BackendAggregation,
 ) -> anyhow::Result<asap_types::PrecomputeMaterialization> {
@@ -3254,12 +3278,10 @@ fn materialization_consumers(
             {
                 continue;
             }
-            let config = aggregation_config_for_materialization(&physical_aggregation(
-                query,
-                &state,
-                query.query_id.clone(),
-                target,
-            ))?;
+            let config = scoped_materialization(
+                &physical_aggregation(query, &state, query.query_id.clone(), target),
+                &state.node,
+            )?;
             consumers
                 .entry(config.policy_fingerprint())
                 .or_default()
@@ -3605,7 +3627,7 @@ mod tests {
     use super::*;
 
     #[test]
-    fn raw_per_entity_state_requires_explicit_additive_reduction() {
+    fn raw_per_entity_state_carries_explicit_isolation() {
         for query in [
             "sum_over_time(m[1m])",
             "quantile_over_time(0.99, m[1m])",
@@ -3618,7 +3640,11 @@ mod tests {
                 .compile(request("per-entity", query), environment)
                 .unwrap();
             assert!(
-                plan.precompute_plan.materializations.is_empty(),
+                plan.precompute_plan
+                    .materializations
+                    .iter()
+                    .all(|state| state.partitioning
+                        == Some(asap_types::sds::PopulationPartitioning::PerEntity)),
                 "{query} pooled source entities"
             );
         }
@@ -4894,8 +4920,8 @@ mod tests {
     }
 
     #[test]
-    fn composable_per_entity_window_delegates_exact_subtree_to_prometheus() {
-        use crate::query_plan::{logical::LogicalOperator, QueryPlanNode};
+    fn composable_per_entity_window_installs_isolated_state() {
+        use crate::query_plan::QueryPlanNode;
         let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
             "../../../docs/examples/asapquery-planning-snapshot.json"
         ))
@@ -4905,24 +4931,16 @@ mod tests {
         entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
         let (request, env) = snapshot.planning_request().unwrap();
         let plan = PhysicalCompiler.compile(request, env).unwrap();
-        assert!(plan.precompute_plan.materializations.is_empty());
+        assert_eq!(plan.precompute_plan.materializations.len(), 1);
+        assert_eq!(
+            plan.precompute_plan.materializations[0].partitioning,
+            Some(asap_types::sds::PopulationPartitioning::PerEntity)
+        );
         let entry = plan.query_plan.lookup("sum_over_time(m[1m])").unwrap();
-        assert!(entry.nodes.values().any(|node| matches!(
-            node,
-            QueryPlanNode::Logical {
-                operator: LogicalOperator::ExactSubquery { query },
-                ..
-            } if query == "sum_over_time(m[1m])"
-        )));
-        assert!(entry.nodes.values().all(|node| !matches!(
-            node,
-            QueryPlanNode::ReadMaterialization { .. }
-                | QueryPlanNode::ExactFallback { .. }
-                | QueryPlanNode::Logical {
-                    operator: LogicalOperator::Scan { .. },
-                    ..
-                }
-        )));
+        assert!(entry
+            .nodes
+            .values()
+            .any(|node| matches!(node, QueryPlanNode::ReadMaterialization { .. })));
     }
 
     // Each operand retains its source and semantic range; a smaller shared pane
@@ -5607,11 +5625,13 @@ mod tests {
             .compile()
             .expect("unquoted v1 compatibility startup remains available");
         let (local, env) = snapshot.clone().planning_request().unwrap();
-        assert!(PhysicalCompiler
-            .compile(local, env)
-            .unwrap_err()
-            .to_string()
-            .contains("native residual substitution requires an exact selected value"));
+        let isolated = PhysicalCompiler.compile(local, env).unwrap();
+        assert!(!isolated.precompute_plan.materializations.is_empty());
+        assert!(isolated
+            .precompute_plan
+            .materializations
+            .iter()
+            .all(|state| state.partitioning.is_some()));
         let (request, environment) = snapshot.planning_request().unwrap();
         let native = crate::physical::workload_cost::with_exact_alternative(request)
             .unwrap()
@@ -5641,11 +5661,13 @@ mod tests {
             .compile()
             .expect("unquoted v1 compatibility startup remains available");
         let (local, env) = snapshot.clone().planning_request().unwrap();
-        assert!(PhysicalCompiler
-            .compile(local, env)
-            .unwrap_err()
-            .to_string()
-            .contains("native residual substitution requires an exact selected value"));
+        let isolated = PhysicalCompiler.compile(local, env).unwrap();
+        assert!(!isolated.precompute_plan.materializations.is_empty());
+        assert!(isolated
+            .precompute_plan
+            .materializations
+            .iter()
+            .all(|state| state.partitioning.is_some()));
         let (request, environment) = snapshot.planning_request().unwrap();
         let native = crate::physical::workload_cost::with_exact_alternative(request)
             .unwrap()
diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs
index 8cff067df..a4453ebc7 100644
--- a/crates/asap_types/src/precompute_plan.rs
+++ b/crates/asap_types/src/precompute_plan.rs
@@ -343,6 +343,49 @@ impl PrecomputePlan {
             installed
                 .validate()
                 .map_err(PrecomputePlanError::CatalogContract)?;
+            let dag = installed
+                .document
+                .decode()
+                .map_err(PrecomputePlanError::CatalogContract)?;
+            for node in &dag.nodes {
+                let Some(crate::executable_plan::BackendNodeBinding::Materialization {
+                    summary_definition,
+                }) = installed.binding.node(node.id)
+                else {
+                    continue;
+                };
+                let Some(config) = self
+                    .materializations
+                    .iter()
+                    .find(|config| config.policy_fingerprint() == summary_definition.fingerprint())
+                else {
+                    return Err(PrecomputePlanError::CatalogContract(
+                        "DAG materialization has no runtime configuration".into(),
+                    ));
+                };
+                if let Some(partitioning) = config.partitioning {
+                    if let planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg {
+                        reduction,
+                        ..
+                    } = &node.payload
+                    {
+                        let expected = match reduction {
+                            planner_types::pre_asap::Reduction::PerEntity => {
+                                crate::sds::PopulationPartitioning::PerEntity
+                            }
+                            planner_types::pre_asap::Reduction::Reduce(_) => {
+                                crate::sds::PopulationPartitioning::Grouped
+                            }
+                        };
+                        if partitioning != expected {
+                            return Err(PrecomputePlanError::CatalogContract(
+                                "runtime population partition disagrees with Planner reduction"
+                                    .into(),
+                            ));
+                        }
+                    }
+                }
+            }
         }
         let mut materializations = BTreeSet::new();
         for materialization in &self.materializations {
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index ad7ad5ac6..58e336257 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -560,13 +560,16 @@ fn route_messages(
             // accumulator per source series. Their emitted label values still
             // follow the physical grouping, so query-time Reduce nodes can
             // combine those independent SDS instances safely.
-            let series_scoped = matches!(
-                config.aggregation_type,
-                asap_types::AggregationType::Increase
-                    | asap_types::AggregationType::MultipleIncrease
-                    | asap_types::AggregationType::MinMax
-                    | asap_types::AggregationType::MultipleMinMax
-            );
+            let series_scoped = config.partitioning
+                == Some(asap_types::sds::PopulationPartitioning::PerEntity)
+                || (config.partitioning.is_none()
+                    && matches!(
+                        config.aggregation_type,
+                        asap_types::AggregationType::Increase
+                            | asap_types::AggregationType::MultipleIncrease
+                            | asap_types::AggregationType::MinMax
+                            | asap_types::AggregationType::MultipleMinMax
+                    ));
             let grouping_pairs: Vec<(&str, &str)> = if series_scoped {
                 Vec::new()
             } else {

From 80e854489c54bd73c4e5f74102fc3559220023d7 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 17:56:24 -0600
Subject: [PATCH 3/8] test: initialize explicit population scope in runtime
 fixtures

---
 .../drivers/ingest/prometheus_remote_write.rs |   2 +
 data_plane/src/drivers/query/servers/http.rs  |   1 +
 .../src/precompute_engine/output_sink.rs      |   1 +
 .../sketch_db/lifecycle/eviction.rs           |   1 +
 .../tests/test_utilities/engine_factories.rs  |   8 +
 .../asapquery_compatibility_process_e2e.rs    |   3 +
 .../tests/support/erp_planning_process.rs     | 247 ++++++++++++++++++
 7 files changed, 263 insertions(+)
 create mode 100644 data_plane/tests/support/erp_planning_process.rs

diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index 58e336257..f01a2f1f7 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -861,6 +861,7 @@ mod tests {
             num_aggregates_to_retain: None,
             table_name: None,
             value_column: None,
+            partitioning: None,
         };
         let policy_fp = aggregation.policy_fp_u64();
         let streaming = StreamingConfig::new(HashMap::from([(policy_fp, aggregation)]));
@@ -914,6 +915,7 @@ mod tests {
                 num_aggregates_to_retain: Some(80),
                 table_name: None,
                 value_column: None,
+                partitioning: None,
             };
         let cms = config(
             AggregationType::CountMinSketchWithHeap,
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index f7756cfda..8db2cabee 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -3673,6 +3673,7 @@ aggregations:
                 num_aggregates_to_retain: None,
                 table_name: None,
                 value_column: None,
+                partitioning: None,
             };
             // PR 5: streaming-config is keyed on the policy
             // fingerprint. Build a marker→fingerprint map so the test
diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs
index bbc112c0a..e124f89cd 100644
--- a/data_plane/src/precompute_engine/output_sink.rs
+++ b/data_plane/src/precompute_engine/output_sink.rs
@@ -344,6 +344,7 @@ mod tests {
             num_aggregates_to_retain: None,
             table_name: None,
             value_column: None,
+            partitioning: None,
         }
     }
 
diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs
index a5feb2f29..183588653 100644
--- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs
+++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs
@@ -275,6 +275,7 @@ mod tests {
             num_aggregates_to_retain: None,
             table_name: None,
             value_column: None,
+            partitioning: None,
         }
     }
 
diff --git a/data_plane/src/tests/test_utilities/engine_factories.rs b/data_plane/src/tests/test_utilities/engine_factories.rs
index b820bdc63..b075a1b50 100644
--- a/data_plane/src/tests/test_utilities/engine_factories.rs
+++ b/data_plane/src/tests/test_utilities/engine_factories.rs
@@ -108,6 +108,7 @@ pub fn create_engine_single_pop_with_aggregated(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        partitioning: None,
     };
     let agg_id = agg_config.policy_fp_u64();
     aggregation_configs.insert(agg_id, agg_config);
@@ -202,6 +203,7 @@ pub fn create_engine_dual_input(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        partitioning: None,
     };
     let value_id = value_agg_config.policy_fp_u64();
     aggregation_configs.insert(value_id, value_agg_config);
@@ -226,6 +228,7 @@ pub fn create_engine_dual_input(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        partitioning: None,
     };
     let keys_id = keys_agg_config.policy_fp_u64();
     aggregation_configs.insert(keys_id, keys_agg_config);
@@ -315,6 +318,7 @@ pub fn create_engine_two_metrics(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        partitioning: None,
     };
     let id_a = agg_config_a.policy_fp_u64();
     aggregation_configs.insert(id_a, agg_config_a);
@@ -338,6 +342,7 @@ pub fn create_engine_two_metrics(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        partitioning: None,
     };
     let id_b = agg_config_b.policy_fp_u64();
     aggregation_configs.insert(id_b, agg_config_b);
@@ -437,6 +442,7 @@ pub fn create_engine_three_metrics(
             num_aggregates_to_retain: None,
             table_name: None,
             value_column: None,
+            partitioning: None,
         };
         let id = cfg.policy_fp_u64();
         ids.push(id);
@@ -513,6 +519,7 @@ pub fn create_engine_multi_timestamp(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        partitioning: None,
     };
     let agg_id = agg_config.policy_fp_u64();
     aggregation_configs.insert(agg_id, agg_config);
@@ -581,6 +588,7 @@ pub fn create_engine_multi_timestamp_with_window(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        partitioning: None,
     };
     let agg_id = agg_config.policy_fp_u64();
     aggregation_configs.insert(agg_id, agg_config);
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index e1c994a0d..ad2d4e3a9 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -18,6 +18,9 @@ use prost::Message;
 use serde_json::Value;
 use tokio::sync::Mutex;
 
+#[path = "support/erp_planning_process.rs"]
+mod erp_planning_process;
+
 struct ChildGuard(Child);
 
 impl Drop for ChildGuard {
diff --git a/data_plane/tests/support/erp_planning_process.rs b/data_plane/tests/support/erp_planning_process.rs
new file mode 100644
index 000000000..4e52bee01
--- /dev/null
+++ b/data_plane/tests/support/erp_planning_process.rs
@@ -0,0 +1,247 @@
+use super::*;
+use control_plane::physical::{compiler::BackendLocalPlanningSnapshot, erp::ErpShapeObserver};
+
+fn measured_profiles(raw: &[f64]) -> Value {
+    let mut records = Vec::new();
+    for k in [32, 128] {
+        let mut error = 0.0f64;
+        let mut bytes = 0usize;
+        for seed in 0..10 {
+            let mut sketch = asap_sketchlib::KllSketch::with_seed(k, seed);
+            for value in raw {
+                sketch.update(*value);
+            }
+            bytes = bytes.max(sketch.sketch_bytes().len());
+            for q in 1..100 {
+                let q = q as f64 / 100.0;
+                let estimate = sketch.quantile(q);
+                let lower = raw.iter().filter(|v| **v < estimate).count() as f64 / raw.len() as f64;
+                let upper =
+                    raw.iter().filter(|v| **v <= estimate).count() as f64 / raw.len() as f64;
+                error = error.max((lower - q).max(q - upper).max(0.0));
+            }
+        }
+        records.push(serde_json::json!({
+            "id": format!("process-test-k{k}"), "sketch": "kll-percall", "implementation": "lib",
+            "parameters": {"k": k}, "trials": 10,
+            "distribution": {"erp_shape": {"family": "zipf", "parameters": {"exponent": 1.0},
+                "cardinality": 16, "benchmark_events": raw.len()}},
+            "error_metrics": {"max_rank_err": error},
+            "resources": {"memory_bytes": bytes, "update_cpu_seconds": 0.0,
+                "query_cpu_seconds": 0.0, "merge_cpu_seconds": 0.0}
+        }));
+    }
+    // This correctness fixture measures error and retained serialized state.
+    // CPU is not the objective or a performance claim in this process test.
+    serde_json::json!({"schema_version": 1,
+        "producer_version": "process-test-measured-error-and-serialized-state-cpu-not-measured",
+        "records": records})
+}
+
+#[tokio::test]
+async fn observed_shape_selects_installed_parameters_and_executes_remote_write() {
+    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("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/-/healthy", get(|| async { "healthy" })),
+        )
+        .await
+        .unwrap();
+    });
+    const QUERY: &str = "quantile_over_time(0.9, erp_latency[5s])";
+    let training: Vec<f64> = (1..=16)
+        .flat_map(|value| std::iter::repeat_n(value as f64, 512 / value))
+        .collect();
+    let raw: Vec<f64> = (1..=16)
+        .rev()
+        .flat_map(|value| std::iter::repeat_n(value as f64, 256 / value))
+        .collect();
+    let mut observer = ErpShapeObserver::new(16).unwrap();
+    for (index, value) in raw.iter().enumerate() {
+        observer.observe(&value.to_string(), index / 100).unwrap();
+    }
+    let observation = observer.snapshot().unwrap();
+    let artifact = measured_profiles(&training);
+    let mut chosen = Vec::new();
+    for only_large in [false, true] {
+        let mut evidence = artifact.clone();
+        if only_large {
+            evidence["records"]
+                .as_array_mut()
+                .unwrap()
+                .retain(|row| row["parameters"]["k"] == 128);
+        }
+        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.2}});
+        fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]);
+        fixture["implementation"]["erp"] = serde_json::json!({
+            "distribution": {"workload": {"external": {"dataset": "held-out-process-stream"}}},
+            "artifact": evidence, "implementation": "lib", "error_metric": "max_rank_err",
+            "min_trials": 10, "expected_updates": raw.len(), "expected_queries": 10.0,
+            "expected_merges": 0.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.1, "max_goodness_of_fit": 0.1,
+                "minimum_confidence": 0.8, "minimum_confidence_margin": 0.05},
+            "runtime": {"allowed_algorithms": ["Kll"], "max_memory_bytes": null}
+        });
+        let policy: control_plane::physical::erp::ErpPlanningInput =
+            serde_json::from_value(fixture["implementation"]["erp"].clone()).unwrap();
+        assert!(matches!(
+            policy.select(
+                planner_types::post_asap::SketchAlgorithm::Kll,
+                0.2,
+                planner_types::post_asap::SketchParams::Kll { k: 128 }
+            ),
+            control_plane::physical::erp::ErpParameterDecision::Empirical { .. }
+        ));
+        let snapshot: BackendLocalPlanningSnapshot =
+            serde_json::from_value(fixture.clone()).unwrap();
+        let plan = snapshot.compile().unwrap();
+        assert_eq!(
+            plan.precompute_plan.materializations.len(),
+            1,
+            "plan={plan:#?}; observation={observation:#?}; evidence={artifact}"
+        );
+        let expected_k = if only_large { 128 } else { 32 };
+        assert_eq!(
+            plan.precompute_plan.materializations[0].parameters["k"],
+            expected_k
+        );
+        chosen.push(plan.precompute_plan.materializations[0].policy_fingerprint());
+        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"))
+                .env("RUST_LOG", "data_plane=debug")
+                .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([
+                    "--precompute-allowed-lateness-ms",
+                    "0",
+                    "--precompute-flush-interval-ms",
+                    "25",
+                ])
+                .stdout(Stdio::inherit())
+                .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 config: Value = client
+            .get(format!("{backend}/api/v1/physical-plan/status"))
+            .send()
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        let config_text = serde_json::to_string(&config).unwrap();
+        assert!(
+            config_text.contains(&chosen.last().unwrap().0.to_string()),
+            "installed ERP identity missing: {config}"
+        );
+        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 samples: Vec<_> = raw
+            .iter()
+            .enumerate()
+            .map(|(i, value)| (base + 1 + i as i64, *value))
+            .collect();
+        assert_eq!(
+            remote_write(
+                &client,
+                &backend,
+                &WriteRequest {
+                    timeseries: vec![
+                        series_with_labels("erp_latency", &[("instance", "a")], &samples),
+                        series_with_labels(
+                            "erp_latency",
+                            &[("instance", "b")],
+                            &samples
+                                .iter()
+                                .map(|(t, v)| (*t, *v + 1000.0))
+                                .collect::<Vec<_>>()
+                        ),
+                    ]
+                }
+            )
+            .await,
+            204
+        );
+        assert_eq!(
+            remote_write(
+                &client,
+                &backend,
+                &WriteRequest {
+                    timeseries: vec![
+                        series_with_labels(
+                            "erp_latency",
+                            &[("instance", "a")],
+                            &[(base + 15001, 1.0)]
+                        ),
+                        series_with_labels(
+                            "erp_latency",
+                            &[("instance", "b")],
+                            &[(base + 15001, 1001.0)]
+                        ),
+                    ]
+                }
+            )
+            .await,
+            204
+        );
+        let result = wait_for_warm_instant(
+            &client,
+            &backend,
+            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,
+            "per-series KLL states must not pool: {result}"
+        );
+        for (instance, offset) in [("a", 0.0), ("b", 1000.0)] {
+            let row = rows
+                .iter()
+                .find(|row| row["metric"]["instance"] == instance)
+                .expect("source labels retained");
+            let estimate = row["value"][1].as_str().unwrap().parse::<f64>().unwrap() - offset;
+            let lower = raw.iter().filter(|v| **v < estimate).count() as f64 / raw.len() as f64;
+            let upper = raw.iter().filter(|v| **v <= estimate).count() as f64 / raw.len() as f64;
+            assert!((lower - 0.9).max(0.9 - upper).max(0.0) <= 0.2, "{result}");
+        }
+    }
+    fallback_task.abort();
+    assert_ne!(
+        chosen[0], chosen[1],
+        "changed evidence must change installed state identity"
+    );
+}

From 2345146ec2f33b56fc45bcd71f19bdfff74faa34 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 17:59:27 -0600
Subject: [PATCH 4/8] test: verify bound population isolation and ERP process
 provenance

---
 control_plane/src/physical/compiler.rs        | 57 +++++++++++++++++--
 .../drivers/ingest/prometheus_remote_write.rs | 26 ++++++++-
 .../tests/support/erp_planning_process.rs     | 15 +++++
 docs/design_docs/shape-aware-erp-v1.md        | 18 ++++++
 4 files changed, 110 insertions(+), 6 deletions(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index cf459fd41..06dde88cf 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -1489,8 +1489,9 @@ impl BackendLocalPlanningSnapshot {
     }
 }
 
-/// Raw accumulators do not retain arbitrary source labels. Preserve native semantics
-/// unless the selected DAG explicitly authorizes pooling the source entities.
+/// Admit per-entity raw state only when the installed partition contract and
+/// scalar input evaluator preserve source rows. Composite updates still require
+/// an executable maintenance evaluator; a partition flag cannot authorize them.
 fn has_unsafe_raw_entity_leaf(
     node: &Rc<SummaryNode>,
     selected: &[Rc<SummaryNode>],
@@ -3223,6 +3224,9 @@ fn scoped_materialization(
     node: &SummaryNode,
 ) -> anyhow::Result<asap_types::PrecomputeMaterialization> {
     let mut config = aggregation_config_for_materialization(aggregation)?;
+    if !matches!(aggregation.aggregation_input, AggregationInput::Raw) {
+        return Ok(config);
+    }
     let SummaryExpr::SummaryAgg { reduction, .. } = &node.expr else {
         anyhow::bail!("materialization lacks SummaryAgg partition contract");
     };
@@ -3343,8 +3347,9 @@ fn shared_pane_origin_ms(
 /// serialized QueryPlan retains the merge edges. Unsupported operators are
 /// intentionally not traversed: QueryPlan lowers them to an explicit exact
 /// fallback node and no unused warm state is provisioned.
-/// Raw accumulators do not retain arbitrary source labels. Preserve native semantics
-/// unless the selected DAG explicitly authorizes pooling the source entities.
+/// Admit per-entity raw state only when the installed partition contract and
+/// scalar input evaluator preserve source rows. Composite updates still require
+/// an executable maintenance evaluator; a partition flag cannot authorize them.
 fn collect_selected_materializations(
     node: &Rc<SummaryNode>,
     composable: bool,
@@ -3626,6 +3631,50 @@ fn stable_workload_plan_id(
 mod tests {
     use super::*;
 
+    #[test]
+    fn installed_partition_must_match_the_bound_dag_reduction() {
+        let mut env = environment(10_000);
+        env.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+        env.collector_ids.clear();
+        let mut plan = PhysicalCompiler
+            .compile(request("scope", "sum_over_time(m[1m])"), env)
+            .unwrap();
+        let installed = plan
+            .precompute_plan
+            .executable_dags
+            .values_mut()
+            .next()
+            .unwrap();
+        let mut dag = installed.document.decode().unwrap();
+        let node = dag
+            .nodes
+            .iter_mut()
+            .find(|node| {
+                matches!(
+                    node.payload,
+                    planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg { .. }
+                )
+            })
+            .unwrap();
+        if let planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg {
+            reduction, ..
+        } = &mut node.payload
+        {
+            *reduction = planner_types::pre_asap::Reduction::by(vec![]);
+        }
+        installed.document = asap_types::executable_plan::OwnedPostAsapDag::from_executable(
+            installed.document.query_id.clone(),
+            &dag,
+        )
+        .unwrap();
+        assert!(plan
+            .precompute_plan
+            .validate()
+            .unwrap_err()
+            .to_string()
+            .contains("partition"));
+    }
+
     #[test]
     fn raw_per_entity_state_carries_explicit_isolation() {
         for query in [
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index f01a2f1f7..1596ab93f 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -923,10 +923,21 @@ mod tests {
             vec!["job".into()],
         );
         let counter = config(AggregationType::Increase, vec!["job".into()], vec![]);
+        let mut kll = config(AggregationType::DatasketchesKLL, vec![], vec![]);
+        kll.partitioning = Some(asap_types::sds::PopulationPartitioning::PerEntity);
+        let kll_fp = kll.policy_fingerprint();
+        let mut pooled_kll = kll.clone();
+        pooled_kll.partitioning = Some(asap_types::sds::PopulationPartitioning::Grouped);
+        let pooled_kll_fp = pooled_kll.policy_fingerprint();
+        assert_ne!(kll_fp, pooled_kll_fp);
         let cms_fp = cms.policy_fingerprint();
         let counter_fp = counter.policy_fingerprint();
-        let streaming =
-            StreamingConfig::new(HashMap::from([(cms_fp.0, cms), (counter_fp.0, counter)]));
+        let streaming = StreamingConfig::new(HashMap::from([
+            (cms_fp.0, cms),
+            (counter_fp.0, counter),
+            (kll_fp.0, kll),
+            (pooled_kll_fp.0, pooled_kll),
+        ]));
         let hot_reload = physical_config(streaming);
         let physical_plan = hot_reload.physical_plan_snapshot().unwrap();
         let (sender, _worker) = mpsc::channel(8);
@@ -976,6 +987,8 @@ mod tests {
         let mut cms_buckets = 0;
         let mut counter_buckets = 0;
         let mut cms_samples = 0;
+        let mut kll_buckets = 0;
+        let mut pooled_kll_buckets = 0;
         for message in messages {
             let WorkerMessage::GroupSamples {
                 policy_fp, samples, ..
@@ -988,8 +1001,17 @@ mod tests {
                 cms_samples += samples.len();
             } else if policy_fp == counter_fp {
                 counter_buckets += 1;
+            } else if policy_fp == kll_fp {
+                kll_buckets += 1;
+            } else if policy_fp == pooled_kll_fp {
+                pooled_kll_buckets += 1;
             }
         }
+        assert_eq!(kll_buckets, 10, "PerEntity KLL keeps every source series");
+        assert_eq!(
+            pooled_kll_buckets, 1,
+            "Grouped empty keys intentionally pool"
+        );
         assert_eq!(cms_buckets, 1, "Reduce([]) has one global CMS SID");
         assert_eq!(cms_samples, 10, "global CMS receives every source series");
         assert_eq!(
diff --git a/data_plane/tests/support/erp_planning_process.rs b/data_plane/tests/support/erp_planning_process.rs
index 4e52bee01..b321fba2b 100644
--- a/data_plane/tests/support/erp_planning_process.rs
+++ b/data_plane/tests/support/erp_planning_process.rs
@@ -116,6 +116,17 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write()
             expected_k
         );
         chosen.push(plan.precompute_plan.materializations[0].policy_fingerprint());
+        eprintln!(
+            "ERP_PLANNED {}",
+            serde_json::json!({
+                "query": QUERY, "available_profiles": policy.artifact.records,
+                "observation": policy.observed_shape,
+                "selected_parameters": plan.precompute_plan.materializations[0].parameters,
+                "materialization": chosen.last(),
+                "partitioning": plan.precompute_plan.materializations[0].partitioning,
+                "query_plan": plan.query_plan,
+            })
+        );
         let output = tempfile::tempdir().unwrap();
         let path = output.path().join("planning.json");
         std::fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap();
@@ -228,6 +239,10 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write()
             2,
             "per-series KLL states must not pool: {result}"
         );
+        eprintln!(
+            "ERP_WARM {}",
+            serde_json::json!({"materialization": chosen.last(), "result": result})
+        );
         for (instance, offset) in [("a", 0.0), ("b", 1000.0)] {
             let row = rows
                 .iter()
diff --git a/docs/design_docs/shape-aware-erp-v1.md b/docs/design_docs/shape-aware-erp-v1.md
index 04a8951b0..5d04d99af 100644
--- a/docs/design_docs/shape-aware-erp-v1.md
+++ b/docs/design_docs/shape-aware-erp-v1.md
@@ -57,3 +57,21 @@ numeric-distribution observation. Equal frequencies produce the canonical
 uniform fit only: Zipf exponent zero describes the same distribution and must
 not create a false ambiguity. Near-uniform, genuinely distinct fits still pass
 through the normal ambiguity policy.
+
+### Population isolation in backend-local execution
+
+A temporal scalar summary has one state per source series when its selected
+`SummaryAgg` uses `Reduction::PerEntity`. An explicit reduction with no grouping
+keys has one pooled population. These are different materializations even when
+source, sketch parameters, and the visible grouping-key list are identical.
+
+The compiler records shared `PopulationPartitioning` metadata in the runtime
+configuration and DataDescriptor. Both identities include the partitioning;
+installation checks it against the bound Planner DAG. Raw ingestion uses the
+full source labels for per-entity routing and the configured grouping for pooled
+routing. Memory estimates count per-entity states against source cardinality.
+Legacy configurations without this metadata retain their existing routing rules.
+
+This is a source-isolation contract, not permission to skip a maintenance update
+expression. Only already-supported scalar update expressions pass the compiler's
+per-entity admission check; other subDAG updates still require a real evaluator.

From 5d6d0d1d7187515958a3b0ec11ed4b932789ddb6 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 18:00:29 -0600
Subject: [PATCH 5/8] test: record ERP and lifecycle selection evidence

---
 data_plane/tests/support/erp_planning_process.rs | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/data_plane/tests/support/erp_planning_process.rs b/data_plane/tests/support/erp_planning_process.rs
index b321fba2b..8ceccdb46 100644
--- a/data_plane/tests/support/erp_planning_process.rs
+++ b/data_plane/tests/support/erp_planning_process.rs
@@ -120,6 +120,8 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write()
             "ERP_PLANNED {}",
             serde_json::json!({
                 "query": QUERY, "available_profiles": policy.artifact.records,
+            "parameter_decision": format!("{:?}", policy.select(planner_types::post_asap::SketchAlgorithm::Kll, 0.2, planner_types::post_asap::SketchParams::Kll { k: 128 })),
+            "lifecycle_estimates": plan.lifecycle_estimates,
                 "observation": policy.observed_shape,
                 "selected_parameters": plan.precompute_plan.materializations[0].parameters,
                 "materialization": chosen.last(),

From bb5149a7fc0ecd9fa9a3e93b5822ecf57f8a5f92 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 18:11:03 -0600
Subject: [PATCH 6/8] fix: validate stored state extent and await finite
 ingestion completion

---
 control_plane/src/physical/compiler.rs        |  8 +--
 crates/asap_types/src/aggregation_config.rs   |  9 +++
 data_plane/src/drivers/query/servers/http.rs  |  2 +-
 .../asapquery_compatibility_process_e2e.rs    | 65 +++++++++----------
 .../tests/support/erp_planning_process.rs     |  4 +-
 docs/developer_docs/erp-process-validation.md | 49 ++++++++++++++
 6 files changed, 93 insertions(+), 44 deletions(-)
 create mode 100644 docs/developer_docs/erp-process-validation.md

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 06dde88cf..f721ed408 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -2162,13 +2162,7 @@ impl PhysicalCompiler {
                         .materializations
                         .iter()
                         .find(|candidate| candidate.policy_fingerprint() == fingerprint)
-                        .map(|candidate| match &candidate.window_layout {
-                            asap_types::WindowMaterializationLayout::FullWindow => {
-                                candidate.window_size
-                            }
-                            layout => layout.base_pane_secs(),
-                        }
-                        .saturating_mul(1_000))
+                        .map(asap_types::PrecomputeMaterialization::stored_window_ms)
                         .ok_or_else(|| {
                             crate::query_plan::QueryPlanError::Invalid(format!(
                                 "compiled binding {} has no precompute materialization",
diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs
index d94dbf145..f1ef3a23c 100644
--- a/crates/asap_types/src/aggregation_config.rs
+++ b/crates/asap_types/src/aggregation_config.rs
@@ -164,6 +164,15 @@ impl AggregationIdInfo {
 pub type AggregationConfig = PrecomputeMaterialization;
 
 impl PrecomputeMaterialization {
+    /// Temporal extent of one stored base state, independent of emission cadence.
+    pub fn stored_window_ms(&self) -> u64 {
+        match &self.window_layout {
+            WindowMaterializationLayout::FullWindow => self.window_size,
+            layout => layout.base_pane_secs(),
+        }
+        .saturating_mul(1_000)
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         aggregation_type: AggregationType,
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index 8db2cabee..3ec38ef38 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -6070,7 +6070,7 @@ pub fn build_active_physical_plan(
                 .iter()
                 .find(|config| config.policy_fingerprint() == binding.materialization.fingerprint())
                 .ok_or_else(|| "query binding has no precompute definition".to_string())?;
-            if binding.window_ms != materialization.slide_interval.saturating_mul(1_000) {
+            if binding.window_ms != materialization.stored_window_ms() {
                 return Err(
                     "query physical pane duration differs from installed precompute definition"
                         .into(),
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index ad2d4e3a9..a9e2c1ffa 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -97,6 +97,19 @@ async fn remote_write(client: &reqwest::Client, base: &str, request: &WriteReque
         .as_u16()
 }
 
+async fn drain_precompute(client: &reqwest::Client, backend: &str) {
+    let response = client
+        .post(format!("{backend}/api/v1/precompute/drain"))
+        .send()
+        .await
+        .unwrap();
+    assert!(
+        response.status().is_success(),
+        "drain failed: {}",
+        response.text().await.unwrap()
+    );
+}
+
 fn first_value(response: &Value, field: &str) -> Option<f64> {
     let samples = response["data"]["result"]
         .as_array()?
@@ -567,6 +580,7 @@ async fn registered_temporal_topk(algorithm: planner_types::post_asap::SketchAlg
     };
     assert_eq!(remote_write(&client, &backend, &watermark).await, 204);
     assert_eq!(remote_write(&client, &backend, &samples).await, 204);
+    drain_precompute(&client, &backend).await;
     let timestamp = (base + 5000) as f64 / 1000.0;
     let instant = tokio::time::timeout(Duration::from_secs(30), async {
         loop {
@@ -1235,6 +1249,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     );
     // A normal Prometheus retry must be accepted without changing sketches.
     assert_eq!(remote_write(&client, &backend, &request).await, 204);
+    drain_precompute(&client, &backend).await;
     let corrupt = client
         .post(format!("{backend}/api/v1/write"))
         .header("content-encoding", "snappy")
@@ -1271,40 +1286,22 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         assert_eq!(range["status"], "success", "{query}: {range}");
         assert!(is_warm(&range), "{query}: {range}");
     }
-    // The bare per-series quantile has no producer binding and must forward
-    // the complete request to the exact backend.
-    for query in ["quantile_over_time(0.5, asap_demo_latency_ms[5s])"] {
-        let instant: Value = client
-            .get(format!("{backend}/api/v1/query"))
-            .query(&[
-                ("query", query.to_string()),
-                ("time", first_eval.to_string()),
-            ])
-            .send()
-            .await
-            .unwrap()
-            .json()
-            .await
-            .unwrap();
-        assert_eq!(instant["data"]["result"][0]["metric"]["fallback"], "true");
-        assert!(!is_warm(&instant));
-        let range: Value = client
-            .get(format!("{backend}/api/v1/query_range"))
-            .query(&[
-                ("query", query.to_string()),
-                ("start", first_eval.to_string()),
-                ("end", second_eval.to_string()),
-                ("step", "5".into()),
-            ])
-            .send()
-            .await
-            .unwrap()
-            .json()
-            .await
-            .unwrap();
-        assert_eq!(range["data"]["result"][0]["metric"]["fallback"], "true");
-        assert!(!is_warm(&range));
-    }
+    // Per-series quantile now has a population-isolated producer.
+    let quantile_query = "quantile_over_time(0.5, asap_demo_latency_ms[5s])";
+    let quantile =
+        wait_for_warm_instant(&client, &backend, quantile_query, first_eval, &backend_log).await;
+    assert!(is_warm(&quantile));
+    let quantile_range = wait_for_warm_range(
+        &client,
+        &backend,
+        quantile_query,
+        first_eval,
+        second_eval,
+        5,
+        &backend_log,
+    )
+    .await;
+    assert!(is_warm(&quantile_range));
     let sum = wait_for_warm_instant(
         &client,
         &backend,
diff --git a/data_plane/tests/support/erp_planning_process.rs b/data_plane/tests/support/erp_planning_process.rs
index 8ceccdb46..aa0cf5083 100644
--- a/data_plane/tests/support/erp_planning_process.rs
+++ b/data_plane/tests/support/erp_planning_process.rs
@@ -135,7 +135,6 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write()
         let port = unused_port();
         let mut child = ChildGuard(
             Command::new(env!("CARGO_BIN_EXE_data_plane"))
-                .env("RUST_LOG", "data_plane=debug")
                 .args([
                     "--forward-unsupported-queries",
                     "--prometheus-server",
@@ -153,7 +152,7 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write()
                     "--precompute-flush-interval-ms",
                     "25",
                 ])
-                .stdout(Stdio::inherit())
+                .stdout(Stdio::null())
                 .stderr(Stdio::inherit())
                 .spawn()
                 .unwrap(),
@@ -227,6 +226,7 @@ async fn observed_shape_selects_installed_parameters_and_executes_remote_write()
             .await,
             204
         );
+        drain_precompute(&client, &backend).await;
         let result = wait_for_warm_instant(
             &client,
             &backend,
diff --git a/docs/developer_docs/erp-process-validation.md b/docs/developer_docs/erp-process-validation.md
new file mode 100644
index 000000000..9950e1ea5
--- /dev/null
+++ b/docs/developer_docs/erp-process-validation.md
@@ -0,0 +1,49 @@
+# ERP-selected KLL process validation
+
+The process test in [erp_planning_process.rs](../../data_plane/tests/support/erp_planning_process.rs)
+checks observation → Planner selection → physical installation → raw ingestion →
+ASAP query execution. This is a correctness fixture, not an o11ybench performance
+result or a claim that empirical rank error is a probabilistic bound.
+
+The fixture measures the runtime KLL implementation at `k=32` and `k=128` using
+10 seeds and 99 quantiles. It records maximum observed tie-aware rank error and
+serialized retained state size. The observer fits a held-out stream; Planner
+selects parameters from the available measured profiles. The first run offers
+both profiles; the second offers only the larger profile. No preselected physical
+plan is supplied to the data-plane process.
+
+Both processes compile the startup planning snapshot and accept remote-write
+samples for two source series. Their value domains differ by 1000, making an
+accidental pooled result detectable. A health-only fallback server has no query
+handler. The test requires two labeled warm ASAP results, each within the declared
+rank-error target of 0.2.
+
+One verified run produced:
+
+| Available profiles | Selected k | Materialization ID | Returned values (a, b) |
+| --- | ---: | ---: | --- |
+| 32, 128 | 32 | 17085259479989488410 | 12, 1012 |
+| 128 | 128 | 81115169662305743 | 12, 1012 |
+
+The measured fixture errors were 0.03336 and 0.00496, respectively. The objective
+uses serialized-state byte-seconds; CPU terms are disabled because this fixture
+does not benchmark CPU. These measurements are test evidence, not a production
+ERP artifact distributed for unrelated workloads. The observer describes ranked
+key frequencies, not arbitrary numeric value spacing.
+
+Run:
+
+```sh
+cargo test -p data_plane --test asapquery_compatibility_process_e2e erp_planning_process -- --nocapture
+```
+
+`ERP_PLANNED` records candidate profiles, fitted observation, selected parameters,
+estimated costs, installed identity, and QueryPlan. `ERP_WARM` records the actual
+HTTP result and execution path. The evaluation workspace retains the captured
+run at `/mydata/erp-production-study/process-evidence.json`; this host-local path
+is not a repository fixture.
+
+Separate compiler and adapter tests cover ERP miss → theoretical sizing → exact
+fallback. This process test establishes two ERP hits; it does not claim a live
+online feedback service, general sketch-family accuracy calibration, or latency,
+CPU, and memory improvements over Prometheus.

From 5b70bd85ee7663c9b7dab28c42916da301970032 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 18:21:14 -0600
Subject: [PATCH 7/8] test: drain shared dashboard ingestion before querying

---
 data_plane/tests/asapquery_compatibility_process_e2e.rs | 1 +
 1 file changed, 1 insertion(+)

diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index a9e2c1ffa..502960ae4 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -886,6 +886,7 @@ async fn run_shared_dashboard(multi_pane: bool) {
         };
         assert_eq!(remote_write(&client, &backend, &close_third).await, 204);
     }
+    drain_precompute(&client, &backend).await;
     let evaluation = base + if multi_pane { 10000 } else { 5000 };
     for (query, expected) in [
         (

From cb735763adbd3f3809fa11bedbef3ba6f486cbc5 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 19:31:21 -0600
Subject: [PATCH 8/8] style: format language-aware scoped materialization call

---
 control_plane/src/physical/compiler.rs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 6aa92948b..0e99bb4d3 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -3190,7 +3190,8 @@ fn scoped_materialization(
     aggregation: &BackendAggregation,
     node: &SummaryNode,
 ) -> anyhow::Result<asap_types::PrecomputeMaterialization> {
-    let mut config = aggregation_config_for_materialization(aggregation, asap_types::QueryLanguage::PromQl)?;
+    let mut config =
+        aggregation_config_for_materialization(aggregation, asap_types::QueryLanguage::PromQl)?;
     if !matches!(aggregation.aggregation_input, AggregationInput::Raw) {
         return Ok(config);
     }