refactor: separate typed summary implementation from level-1 acceptance

control_plane/src/clickhouse.rs

@@ -389,7 +389,7 @@ fn materialize_selected_sql(
     let aggregation = BackendAggregation {
         aggregation_id: String::new(),
         metric_name: format!("{table}.{}", value.column().unwrap_or("constant")),
-        family: crate::physical::compiler::physical_materialization_family(family),
+        family: family.clone(),
         window_secs,
         spatial_filter: String::new(),
         grouping: grouping.names(),
@@ -610,7 +610,7 @@ fn bind_selected_node(
         ..
     } = clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms)
         .map_err(crate::query_plan::QueryPlanError::Invalid)?;
-    let expected = crate::physical::compiler::physical_materialization_family(family);
+    let expected = family.clone();
     let selected = select_materialization(
         &request.precompute_plan.materializations,
         &table_ref,

control_plane/src/emit/backend_wire.rs

@@ -5,7 +5,7 @@
 //! * the storage-routing table, which maps each metric's materialized summary
 //!   families to the query shapes the ASAP tier serves natively versus the
 //!   ones that belong to the archive;
-//! * the aggregation and readout JSON the backend's `AggregationConfig`
+//! * the aggregation and readout JSON the backend's `PrecomputeMaterialization`
 //!   parser consumes.
 //!
 //! `backend_plan::from_stage_config` reuses [`build_backend_aggregation_json`]

control_plane/src/emit/mod.rs

@@ -1,7 +1,7 @@
 //! Backend-facing emission for a compiled physical plan.
 //!
 //! * [`backend_wire`] builds the storage-routing table and the aggregation /
-//!   readout JSON the backend's `AggregationConfig` parser consumes.
+//!   readout JSON the backend's `PrecomputeMaterialization` parser consumes.
 //! * [`monitor`] carries the CDM monitor declarations.
 
 pub mod backend_wire;

control_plane/src/physical/backend_stage.rs

@@ -35,7 +35,7 @@ pub struct BackendAggregation {
     /// Internal-only id (see struct doc). Not on the wire.
     pub aggregation_id: String,
     /// Source metric the aggregation runs over. Required by the backend's
-    /// `AggregationConfig` parser.
+    /// `PrecomputeMaterialization` parser.
     pub metric_name: String,
     /// Planner-owned committed summary identity. Sketch entries carry a
     /// validated `SketchKind` (category + algorithm + params); exact entries

control_plane/src/physical/compiler.rs

@@ -1256,10 +1256,7 @@ impl PhysicalPlanCompiler {
                     .with_window_implementation_costs(window_costs);
                 let metric = selected.metric.clone();
                 let aggregation_id = format!("{}:{ordinal}:{}", query.query_id, metric);
-                // Rate is a readout over the same reset-aware counter state
-                // as Increase. Keep that semantic distinction in QueryPlan,
-                // while the physical store binds both to Increase state.
-                let physical_family = physical_materialization_family(&selected.family);
+                let physical_family = selected.family.clone();
                 let physical_algorithm = match &physical_family {
                     SummaryFamilyType::ExactAggregate(kind, _) => {
                         format!("{kind:?}").to_ascii_lowercase()
@@ -1698,7 +1695,7 @@ impl PhysicalPlanCompiler {
                         .map_err(|error| crate::query_plan::QueryPlanError::Invalid(error.to_string()))?
                         .family;
                     let window_ms = materialization.window_size.saturating_mul(1_000);
-                    if materialization_family != physical_materialization_family(node_family)
+                    if materialization_family != *node_family
                         || window_ms == 0
                         || source_window.unwrap_or(query.query_lookback_seconds).saturating_mul(1_000)
                             % window_ms != 0
@@ -2843,13 +2840,11 @@ pub(super) fn estimated_state_bytes(
         A::HLL => 1u128 << parameter(&["precision", "p"], 14).min(24),
         A::DDSketch => 64 * 1024,
         A::Sum
+        | A::Count
         | A::Increase
+        | A::Rate
         | A::Min
         | A::Max
-        | A::MultipleSum
-        | A::MultipleIncrease
-        | A::MultipleMin
-        | A::MultipleMax
         | A::SingleSubpopulation
         | A::MultipleSubpopulation => 256,
     }
@@ -2867,7 +2862,7 @@ fn retained_partition_count(
     if materialization.partitioning == Some(asap_types::sds::PopulationPartitioning::PerEntity)
         || matches!(
             materialization.aggregation_type,
-            A::Increase | A::MultipleIncrease | A::Min | A::Max | A::MultipleMin | A::MultipleMax
+            A::Increase | A::Rate | A::Min | A::Max
         )
         || !materialization.grouping_labels.names().is_empty()
     {
@@ -3114,7 +3109,7 @@ pub(crate) fn raw_materialization_input_contract(
     )
 }
 
-fn raw_time_series_input_contract(
+pub fn raw_time_series_input_contract(
     expr: &QueryExpr,
     exact: bool,
 ) -> Result<(String, Option<u64>, String), String> {
@@ -3310,7 +3305,7 @@ fn physical_aggregation(
     BackendAggregation {
         aggregation_id,
         metric_name: selected.metric.clone(),
-        family: physical_materialization_family(&selected.family),
+        family: selected.family.clone(),
         window_secs: selected.window_secs.unwrap_or(query.query_lookback_seconds),
         spatial_filter: selected.spatial_filter.clone(),
         grouping: selected
@@ -3720,26 +3715,6 @@ fn collect_selected_materializations(
     Ok(selected)
 }
 
-pub(crate) fn physical_materialization_family(family: &SummaryFamilyType) -> SummaryFamilyType {
-    match family {
-        SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Count, _) => {
-            // The SummaryStore Sum accumulator retains the observation count
-            // alongside its sum. Both logical states can share this producer.
-            SummaryFamilyType::ExactAggregate(
-                planner_types::post_asap::ExactKind::Sum,
-                planner_types::post_asap::ExactParams::Sum,
-            )
-        }
-        SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Rate, _) => {
-            SummaryFamilyType::ExactAggregate(
-                planner_types::post_asap::ExactKind::Increase,
-                planner_types::post_asap::ExactParams::Increase,
-            )
-        }
-        _ => family.clone(),
-    }
-}
-
 pub(super) fn sketch_params_json(params: &planner_types::post_asap::SketchParams) -> Value {
     use planner_types::post_asap::SketchParams as P;
     match params {
@@ -4529,8 +4504,7 @@ pub(crate) mod tests {
             .find(|materialization| {
                 matches!(
                     materialization.aggregation_type,
-                    asap_types::AggregationType::Increase
-                        | asap_types::AggregationType::MultipleIncrease
+                    asap_types::AggregationType::Rate
                 )
             })
             .expect("reset-aware exact counter");
@@ -5101,8 +5075,7 @@ pub(crate) mod tests {
             .iter()
             .all(|m| !matches!(
                 m.aggregation_type,
-                asap_types::AggregationType::Increase
-                    | asap_types::AggregationType::MultipleIncrease
+                asap_types::AggregationType::Increase | asap_types::AggregationType::Rate
             )));
         let entry = plan.query_plan.entries.values().next().unwrap();
         assert!(!entry.materialization_bindings().is_empty());
@@ -5660,7 +5633,7 @@ pub(crate) mod tests {
     }
 
     #[test]
-    fn rate_and_increase_share_physical_counter_state() {
+    fn rate_and_increase_keep_planner_families_distinct() {
         let mut workload = request("rate", "rate(m[1m])");
         workload
             .queries
@@ -5669,10 +5642,14 @@ pub(crate) mod tests {
             .compile_promql(workload, environment(10_000))
             .unwrap();
         assert_eq!(bundle.query_plan.entries.len(), 2);
-        assert_eq!(bundle.precompute_plan.materializations.len(), 1);
+        assert_eq!(bundle.precompute_plan.materializations.len(), 2);
         for collector in &bundle.collector_plans {
-            assert_eq!(collector.materializations.len(), 1);
-            assert_eq!(collector.materializations[0].algorithm, "increase");
+            let algorithms: std::collections::BTreeSet<_> = collector
+                .materializations
+                .iter()
+                .map(|materialization| materialization.algorithm.as_str())
+                .collect();
+            assert_eq!(algorithms, ["increase", "rate"].into());
         }
     }
 
@@ -5692,8 +5669,7 @@ pub(crate) mod tests {
     }
 
     #[test]
-    fn exact_dashboard_binds_sum_and_count_to_one_local_producer() {
-        // Both dashboard roots use one packed raw accumulator, with explicit readouts.
+    fn exact_dashboard_preserves_distinct_sum_and_count_producers() {
         let mut snapshot: BackendLocalPlanningInput = serde_json::from_str(include_str!(
             "../../../docs/examples/asapquery-planning-snapshot.json"
         ))
@@ -5709,7 +5685,7 @@ pub(crate) mod tests {
         entries.push(mean);
         let (request, env) = snapshot.into_physical_compilation_request().unwrap();
         let bundle = PhysicalPlanCompiler.compile_promql(request, env).unwrap();
-        assert_eq!(bundle.precompute_plan.materializations.len(), 1);
+        assert_eq!(bundle.precompute_plan.materializations.len(), 2);
         assert_eq!(bundle.query_plan.entries.len(), 2);
         for entry in bundle.query_plan.entries.values() {
             assert!(
@@ -5719,7 +5695,7 @@ pub(crate) mod tests {
                 )),
                 "{entry:?}"
             );
-            assert_eq!(entry.materialization_bindings().len(), 1);
+            assert!(!entry.materialization_bindings().is_empty());
         }
         assert!(bundle
             .query_plan
@@ -7484,8 +7460,8 @@ pub(crate) mod tests {
                 assert_eq!(
                     materialization.accumulator_spec().unwrap().family,
                     SummaryFamilyType::ExactAggregate(
-                        planner_types::post_asap::ExactKind::Increase,
-                        planner_types::post_asap::ExactParams::Increase,
+                        planner_types::post_asap::ExactKind::Rate,
+                        planner_types::post_asap::ExactParams::Rate,
                     )
                 );
             }

control_plane/src/physical/pane_reuse.rs

@@ -34,10 +34,7 @@ pub(super) fn share_additive_panes(
         if !seen.insert(old)
             || m.derived_input.is_some()
             || derived_sources.contains(&old)
-            || !matches!(
-                m.aggregation_type,
-                AggregationType::Sum | AggregationType::MultipleSum
-            )
+            || !matches!(m.aggregation_type, AggregationType::Sum)
         {
             continue;
         }

control_plane/src/physical/post_asap/lower.rs

@@ -1,5 +1,4 @@
 //! Query binding delegates selection to Planner's costed workload search.
-//! Backend-specific rate normalization remains part of the physical binding.
 
 #![allow(dead_code)]
 
@@ -118,41 +117,8 @@ fn bind_recursive(
             ))
         }
 
-        _ => {
-            let rewritten = rewrite_rate_to_increase(expr);
-            let node = crate::planner_selection::select_query(&rewritten, cost_model)?;
-            Ok(PostAsapPlan::Summary(node))
-        }
-    }
-}
-
-/// Rewrite Rate to Increase along the aggregate spine traversed by Planner.
-/// This deployment computes rate by dividing the Increase readout by window
-/// seconds, rather than storing a separate Rate accumulator.
-fn rewrite_rate_to_increase(expr: &QueryExpr) -> QueryExpr {
-    match expr {
-        QueryExpr::Aggregate {
-            reduction,
-            measures: aggs,
-            output_names,
-            having,
-            child,
-        } => QueryExpr::Aggregate {
-            reduction: reduction.clone(),
-            measures: aggs
-                .iter()
-                .map(|intent| {
-                    if matches!(intent, AggIntent::Rate) {
-                        AggIntent::Increase
-                    } else {
-                        intent.clone()
-                    }
-                })
-                .collect(),
-            output_names: output_names.clone(),
-            having: having.clone(),
-            child: Rc::new(rewrite_rate_to_increase(child)),
-        },
-        other => other.clone(),
+        _ => Ok(PostAsapPlan::Summary(
+            crate::planner_selection::select_query(expr, cost_model)?,
+        )),
     }
 }

control_plane/src/physical/post_asap/tests.rs

@@ -474,13 +474,9 @@ fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() {
 /// `ONLY_SPATIAL` — `sum by (host) (m)`.
 /// Control plane path: `Aggregate{Sum, by=[host]}` over a bare `Scan`.
 ///
-/// The old locally-defined `AggregationType::MultipleSum` (keyed vs
-/// unkeyed sum) identity no longer exists at the L4 IR level —
-/// `SummaryKind::Sum` covers both; the keyed/unkeyed distinction now
-/// lives on `SummaryAgg::by` (non-empty ⇒ the old "MultipleSum" shape),
-/// per `emit::mod.rs`'s exact-accumulator classification notes.
+/// Family remains Sum; the reduction carries the grouping columns.
 #[test]
-fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() {
+fn phase_b_pattern_only_spatial_aggregate_binds_to_grouped_sum() {
     let expr = QueryExpr::Aggregate {
         reduction: Reduction::by(vec![1]), // service column
         measures: vec![AggIntent::Sum { col: None }],
@@ -501,7 +497,7 @@ fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() {
                 assert_eq!(
                     reduction.group_keys().map(|k| k.keys()),
                     Some(&[1][..]),
-                    "keyed sum must carry the group-by column (the MultipleSum-equivalent signal)"
+                    "Sum reduction must retain the group-by column"
                 );
             }
             other => panic!("expected SummaryAgg(Sum, by=[1]), got {other:?}"),
@@ -511,14 +507,9 @@ fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() {
 }
 
 /// `ONE_TEMPORAL_ONE_SPATIAL` — `sum by (host) (rate(m[5m]))`.
-/// `bind_query_expr` (not `implement_tree` directly) rewrites
-/// `AggIntent::Rate` to `AggIntent::Increase` before binding (see
-/// `lower.rs`'s `rewrite_rate_to_increase` — this deployment's data
-/// plane has no Rate accumulator). The old
-/// `AggregationType::MultipleIncrease` identity is now
-/// `SummaryKind::Increase` with a non-empty `by`.
+/// Planner preserves the Rate family and the `by` reduction independently.
 #[test]
-fn phase_b_pattern_temporal_and_spatial_combined_binds_to_multiple_increase() {
+fn phase_b_pattern_temporal_and_spatial_combined_preserves_rate() {
     let expr = QueryExpr::Aggregate {
         reduction: Reduction::by(vec![1]),
         measures: vec![AggIntent::Rate],
@@ -534,11 +525,11 @@ fn phase_b_pattern_temporal_and_spatial_combined_binds_to_multiple_increase() {
             } => {
                 assert_eq!(
                     family,
-                    &SummaryFamilyType::ExactAggregate(ExactKind::Increase, ExactParams::Increase)
+                    &SummaryFamilyType::ExactAggregate(ExactKind::Rate, ExactParams::Rate)
                 );
                 assert_eq!(reduction.group_keys().map(|k| k.keys()), Some(&[1][..]));
             }
-            other => panic!("expected SummaryAgg(Increase, by=[1]), got {other:?}"),
+            other => panic!("expected SummaryAgg(Rate, by=[1]), got {other:?}"),
         },
         other => panic!("expected Committed(Summary(_)), got {other:?}"),
     }
@@ -671,12 +662,9 @@ fn phase_b_e2e_sum_by_preserves_grouping_label() {
     );
 }
 
-/// `rate_increase.yaml` — the legacy planner emits a MultipleIncrease
-/// (counter-reset adjusted) row. Control plane path: `Aggregate{Rate}` over
-/// `Window` → `bind_query_expr` rewrites `Rate` to `Increase` and binds an
-/// exact accumulator (`SummaryAgg{Increase}`) — no approximate summary
-/// family. Both paths produce a single non-summary streaming row; the L5
-/// emitter is the one that picks the actual MultipleIncrease processor.
+/// A Rate query keeps Planner's exact Rate family through binding. The
+/// physical emitter chooses the runtime processor without changing that
+/// family identity.
 #[test]
 fn phase_b_e2e_rate_falls_through_to_logical() {
     let bound = pipeline_l1_to_l4(