fix: preserve per-series population in ERP execution

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>],
@@ -1513,9 +1514,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, _))
@@ -1871,10 +1875,8 @@ impl PhysicalCompiler {
                     aggregation_id.clone(),
                     environment.target,
                 );
-                let precompute_materialization = aggregation_config_for_materialization(
-                    &aggregation,
-                    asap_types::QueryLanguage::PromQl,
-                )?;
+                let precompute_materialization =
+                    scoped_materialization(&aggregation, &selected.node)?;
                 let materialization = precompute_materialization.policy_fingerprint();
                 let state_consumers = consumers[&materialization]
                     .iter()
@@ -1905,10 +1907,8 @@ impl PhysicalCompiler {
                 // Preserve semantic window and evaluation cadence independently
                 // from the selected storage representation.
                 aggregation.window_secs = window_implementation.window_secs;
-                let mut runtime_materialization = aggregation_config_for_materialization(
-                    &aggregation,
-                    asap_types::QueryLanguage::PromQl,
-                )?;
+                let mut runtime_materialization =
+                    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 =
@@ -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",
@@ -2836,10 +2830,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 {
@@ -3190,6 +3186,29 @@ 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, asap_types::QueryLanguage::PromQl)?;
+    if !matches!(aggregation.aggregation_input, AggregationInput::Raw) {
+        return Ok(config);
+    }
+    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,
     language: asap_types::QueryLanguage,
@@ -3241,9 +3260,9 @@ fn materialization_consumers(
             {
                 continue;
             }
-            let config = aggregation_config_for_materialization(
+            let config = scoped_materialization(
                 &physical_aggregation(query, &state, query.query_id.clone(), target),
-                asap_types::QueryLanguage::PromQl,
+                &state.node,
             )?;
             consumers
                 .entry(config.policy_fingerprint())
@@ -3306,8 +3325,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,
@@ -3590,7 +3610,51 @@ mod tests {
     use super::*;
 
     #[test]
-    fn raw_per_entity_state_requires_explicit_additive_reduction() {
+    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 [
             "sum_over_time(m[1m])",
             "quantile_over_time(0.99, m[1m])",
@@ -3603,7 +3667,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"
             );
         }
@@ -4879,8 +4947,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"
         ))
@@ -4890,24 +4958,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
@@ -5592,11 +5652,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()
@@ -5626,11 +5688,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()

crates/asap_types/src/aggregation_config.rs

@@ -177,6 +177,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)
+    }
+
     pub fn population_filter_canonical(&self) -> Result<String, String> {
         if let Some(column) = &self.table_timestamp_column {
             if self.table_name.is_none() || column.is_empty() {

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 {