refactor!: compose grouped TopK and separate operator execution phase

crates/asap-aware-mapping/src/accuracy/estimators/mod.rs

@@ -99,7 +99,7 @@ pub(crate) fn size_params(
             SketchParams::CmsWithHeap {
                 width: cms::cms_width(eps),
                 depth: cms::cms_depth(delta),
-                heap_size: k as u32,
+                heap_size: topk_capacity(k, eps),
             }
         }
         // Non-preferred candidates (DDSketch / Theta / Kmv / CountSketch /
@@ -124,11 +124,19 @@ pub(crate) fn size_params(
             SketchParams::CountSketchWithHeap {
                 width: count_sketch::count_sketch_width(eps),
                 depth: count_sketch::count_sketch_depth(delta),
-                heap_size: k as u32,
+                heap_size: topk_capacity(k, eps),
             }
         }
     }
 }
+/// Accuracy-dependent candidate budget, not a completeness theorem. The
+/// membership model must still certify the selected set independently.
+pub(crate) fn topk_capacity(k: usize, eps: f64) -> u32 {
+    u32::try_from(k)
+        .unwrap_or(u32::MAX)
+        .max(saturating_ceil(1.0 / eps, 1, 1 << 26))
+}
+
 /// `⌈x⌉` clamped to `[lo, hi]`; NaN / non-positive x saturate to `hi`
 /// (a degenerate ε means "as accurate as this family goes").
 pub(crate) fn saturating_ceil(x: f64, lo: u32, hi: u32) -> u32 {

crates/asap-aware-mapping/src/accuracy/evidence.rs

@@ -96,6 +96,16 @@ pub trait AccuracyEvidenceProvider {
         None
     }
 
+    /// Enforced upper bound on distinct (partition, item) identities across a
+    /// complete TopK readout. Used to union-bound score errors for adaptively
+    /// selected candidates. Observed cardinality is not sufficient evidence.
+    fn topk_max_distinct_items(
+        &self,
+        _expression: &asap_types::pre_asap::QueryExpr,
+    ) -> Option<u64> {
+        None
+    }
+
     /// Proof scoped to this complete quantile expression, including its source,
     /// filters, grouping and window. `None` means unknown, including emptiness.
     fn quantile_input_domain(

crates/asap-aware-mapping/src/accuracy/mod.rs

@@ -123,6 +123,8 @@ impl AccuracyModel for DefaultAccuracyModel {
     }
 }
 
+pub(crate) use estimators::topk_capacity;
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/asap-aware-mapping/src/cost_model.rs

@@ -88,36 +88,36 @@ pub struct CostProvenance {
 
 /// Which mixed-execution shapes the downstream runtime can actually
 /// execute (issue #171). [`crate::exact_composition::ExactCompositionStrategy`]
-/// proposes an `ValueOperationAtReadTime` candidate only when
-/// `read_time` is set, and an `ValueOperationAtMaintenanceTime` candidate only
-/// when `maintenance_time` is — a runtime that cannot run an exact
+/// proposes an `ValueOperationAtQueryTime` candidate only when
+/// `query_time` is set, and an `ValueOperationAtIngestionTime` candidate only
+/// when `ingestion_time` is — a runtime that cannot run an exact
 /// operator on the update path must never be handed one.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub struct ValueOperationCapabilities {
     /// The runtime can apply an exact operator to summary readouts at
     /// query evaluation time.
-    pub read_time: bool,
+    pub query_time: bool,
     /// The runtime can apply an exact row transform on the update path,
     /// feeding its output into maintained summary state.
-    pub maintenance_time: bool,
+    pub ingestion_time: bool,
 }
 
 impl ValueOperationCapabilities {
     /// Neither shape supported.
     pub const NONE: Self = Self {
-        read_time: false,
-        maintenance_time: false,
+        query_time: false,
+        ingestion_time: false,
     };
     /// Both shapes supported.
     pub const ALL: Self = Self {
-        read_time: true,
-        maintenance_time: true,
+        query_time: true,
+        ingestion_time: true,
     };
 
     pub fn supports(self, placement: OperationPlacement) -> bool {
         match placement {
-            OperationPlacement::Read => self.read_time,
-            OperationPlacement::Maintenance => self.maintenance_time,
+            OperationPlacement::Read => self.query_time,
+            OperationPlacement::Maintenance => self.ingestion_time,
         }
     }
 }

crates/asap-aware-mapping/src/exact_composition.rs

@@ -97,15 +97,15 @@ impl OperationPlacement {
     /// The availability the composed operator consumes and produces.
     pub fn data_state(self) -> ExecutionDataState {
         match self {
-            Self::Read => ExecutionDataState::READ_ROWS,
-            Self::Maintenance => ExecutionDataState::MAINTENANCE_ROWS,
+            Self::Read => ExecutionDataState::QUERY_ROWS,
+            Self::Maintenance => ExecutionDataState::INGESTION_ROWS,
         }
     }
 
     pub fn provenance(self) -> ReplacementProvenance {
         match self {
-            Self::Read => ReplacementProvenance::ValueOperationAtReadTime,
-            Self::Maintenance => ReplacementProvenance::ValueOperationAtMaintenanceTime,
+            Self::Read => ReplacementProvenance::ValueOperationAtQueryTime,
+            Self::Maintenance => ReplacementProvenance::ValueOperationAtIngestionTime,
         }
     }
 }
@@ -199,9 +199,9 @@ impl ExactComposition {
             },
         };
         let timing = match self.placement {
-            OperationPlacement::Read => asap_types::post_asap::ExecutionTiming::ReadTime,
+            OperationPlacement::Read => asap_types::post_asap::ExecutionTiming::QueryTime,
             OperationPlacement::Maintenance => {
-                asap_types::post_asap::ExecutionTiming::MaintenanceTime
+                asap_types::post_asap::ExecutionTiming::IngestionTime
             }
         };
         let expr = SummaryExpr::ValueOperation {
@@ -230,7 +230,7 @@ impl ExactComposition {
 /// Which exact reducers may run as a query-time fold over readout rows.
 /// `Count` only at `Exact` accuracy (an approximate count is a sketch
 /// target, not an exact fold).
-fn is_read_time_reducer(intent: &AggIntent) -> bool {
+fn is_query_time_reducer(intent: &AggIntent) -> bool {
     matches!(
         intent,
         AggIntent::Sum { .. }
@@ -258,7 +258,7 @@ fn needs_readout(implementation: &Realization) -> bool {
 }
 
 /// The `(op, child)` of a read-time operation-shaped target, or `None`.
-fn read_time_shape(
+fn query_time_shape(
     root: &QueryExpr,
     cost_model: &dyn CostModel,
 ) -> Option<(ExactOperation, Rc<QueryExpr>, AggIntent)> {
@@ -281,7 +281,7 @@ fn read_time_shape(
     let [intent] = measures.as_slice() else {
         return None;
     };
-    if !is_read_time_reducer(intent) {
+    if !is_query_time_reducer(intent) {
         return None;
     }
     let child_intent = bindable_intent(child)?;
@@ -308,7 +308,7 @@ fn read_time_shape(
 
 /// The `(op, child)` of a function-shaped target — a per-entity exact
 /// transform with no accumulator form — or `None`.
-fn maintenance_time_shape(
+fn ingestion_time_shape(
     root: &QueryExpr,
     cost_model: &dyn CostModel,
 ) -> Option<(ExactOperation, Rc<QueryExpr>, AggIntent)> {
@@ -382,14 +382,14 @@ impl<'a> ExactCompositionStrategy<'a> {
         let schema = asap_types::post_asap::execution_data_state::lift_plain(&schema);
         let mut out = Vec::new();
 
-        if let Some((op, child, intent)) = read_time_shape(target.root, self.cost_model) {
+        if let Some((op, child, intent)) = query_time_shape(target.root, self.cost_model) {
             if self
                 .cost_model
                 .value_operation_support_evidence(&op, OperationPlacement::Read)
                 != Some(false)
             {
                 let child_desc =
-                    describe_intent(bindable_intent(&child).expect("checked by read_time_shape"));
+                    describe_intent(bindable_intent(&child).expect("checked by query_time_shape"));
                 out.push(ReplacementSubDAG {
                     strategy: "ExactCompositionStrategy",
                     replacement: Replacement::ExactComposition(ExactComposition {
@@ -398,7 +398,7 @@ impl<'a> ExactCompositionStrategy<'a> {
                         child_target: child,
                         schema: schema.clone(),
                     }),
-                    provenance: ReplacementProvenance::ValueOperationAtReadTime,
+                    provenance: ReplacementProvenance::ValueOperationAtQueryTime,
                     rationale: format!(
                         "{} is an exact fold whose input is the readout of {} — a maintained \
                          accumulator cannot consume query-time values, so instead of collapsing \
@@ -412,7 +412,7 @@ impl<'a> ExactCompositionStrategy<'a> {
             }
         }
 
-        if let Some((op, child, intent)) = maintenance_time_shape(target.root, self.cost_model) {
+        if let Some((op, child, intent)) = ingestion_time_shape(target.root, self.cost_model) {
             if self
                 .cost_model
                 .value_operation_support_evidence(&op, OperationPlacement::Maintenance)
@@ -426,7 +426,7 @@ impl<'a> ExactCompositionStrategy<'a> {
                         child_target: child,
                         schema,
                     }),
-                    provenance: ReplacementProvenance::ValueOperationAtMaintenanceTime,
+                    provenance: ReplacementProvenance::ValueOperationAtIngestionTime,
                     rationale: format!(
                         "{} is an exact per-entity function with no accumulator form; as an \
                          explicit ExactMaintenance on the update path its output can feed a \
@@ -505,7 +505,7 @@ mod tests {
     }
 
     #[test]
-    fn proposes_read_time_operation_for_max_over_quantile() {
+    fn proposes_query_time_operation_for_max_over_quantile() {
         let root = max_over_quantile();
         let target = TargetSubDAG::new(&root);
         let strategy = ExactCompositionStrategy::default_cost_model();
@@ -521,7 +521,7 @@ mod tests {
         assert_eq!(comp.placement, OperationPlacement::Read);
         assert_eq!(
             candidates[0].provenance,
-            ReplacementProvenance::ValueOperationAtReadTime
+            ReplacementProvenance::ValueOperationAtQueryTime
         );
         let QueryExpr::Aggregate { child, .. } = root.as_ref() else {
             unreachable!()
@@ -535,7 +535,7 @@ mod tests {
     }
 
     #[test]
-    fn proposes_read_time_operation_for_avg_over_quantile_alongside_the_rewrite() {
+    fn proposes_query_time_operation_for_avg_over_quantile_alongside_the_rewrite() {
         let inner = agg(vec![2], default_quantile(0.99), metric_scan(&["zone"]));
         let root = Rc::new(agg(vec![0], AggIntent::Avg { col: None }, inner));
         let target = TargetSubDAG::new(&root);
@@ -550,14 +550,14 @@ mod tests {
     }
 
     #[test]
-    fn proposes_maintenance_time_operation_for_a_per_entity_pass_through_over_raw_input() {
+    fn proposes_ingestion_time_operation_for_a_per_entity_pass_through_over_raw_input() {
         let root = Rc::new(per_entity(AggIntent::Deriv, metric_scan(&["zone"])));
         let target = TargetSubDAG::new(&root);
         let candidates = ExactCompositionStrategy::default_cost_model().replacements(&target);
         assert_eq!(candidates.len(), 1);
         assert_eq!(
             candidates[0].provenance,
-            ReplacementProvenance::ValueOperationAtMaintenanceTime
+            ReplacementProvenance::ValueOperationAtIngestionTime
         );
     }
 
@@ -607,7 +607,7 @@ mod tests {
     }
 
     #[test]
-    fn compose_rejects_a_maintained_state_child_for_a_read_time_operation() {
+    fn compose_rejects_a_maintained_state_child_for_a_query_time_operation() {
         let root = max_over_quantile();
         let target = TargetSubDAG::new(&root);
         let candidates = ExactCompositionStrategy::default_cost_model().replacements(&target);
@@ -638,7 +638,7 @@ mod tests {
         assert!(matches!(
             composed.expr,
             SummaryExpr::ValueOperation {
-                timing: ExecutionTiming::ReadTime,
+                timing: ExecutionTiming::QueryTime,
                 ..
             }
         ));
@@ -650,7 +650,7 @@ mod tests {
     }
 
     #[test]
-    fn compose_rejects_a_readout_child_for_a_maintenance_time_operation() {
+    fn compose_rejects_a_readout_child_for_a_ingestion_time_operation() {
         let inner = agg(vec![2], default_quantile(0.99), metric_scan(&["zone"]));
         let root = Rc::new(per_entity(AggIntent::Deriv, inner));
         let candidates =
@@ -673,7 +673,7 @@ mod tests {
         assert!(matches!(
             comp.compose(raw).unwrap().expr,
             SummaryExpr::ValueOperation {
-                timing: ExecutionTiming::MaintenanceTime,
+                timing: ExecutionTiming::IngestionTime,
                 ..
             }
         ));

crates/asap-aware-mapping/src/maintained_population.rs

@@ -229,7 +229,7 @@ impl MaintainedPopulationStrategy {
                         cols: cols.clone(),
                         qualifier: qualifier.clone(),
                     },
-                    timing: ExecutionTiming::ReadTime,
+                    timing: ExecutionTiming::QueryTime,
                 },
             }));
         }
@@ -258,7 +258,7 @@ impl MaintainedPopulationStrategy {
             expr: SummaryExpr::ValueOperation {
                 child: scan,
                 operation: ValueOperation::MaintainPopulation { population },
-                timing: ExecutionTiming::MaintenanceTime,
+                timing: ExecutionTiming::IngestionTime,
             },
             schema: input_schema,
             guarantee: Some(ResultGuarantee::exact(
@@ -269,7 +269,7 @@ impl MaintainedPopulationStrategy {
             expr: SummaryExpr::ValueOperation {
                 child: maintained,
                 operation: ValueOperation::ReadPopulation { readout },
-                timing: ExecutionTiming::ReadTime,
+                timing: ExecutionTiming::QueryTime,
             },
             schema: plain(root.output_schema().ok()?),
             guarantee: Some(ResultGuarantee::exact("exact current-population readout")),