Merge pull request #335 from ProjectASAP/feat/cost-streaming-summary-dag

feat(cost): cost complete streaming summary DAGs

Author: zzylol

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

@@ -1948,7 +1948,7 @@ pub enum AnalyticalCostError {
     IncompatibleLifecycleGuarantee,
     #[error("bootstrap row and byte evidence must either both be zero or both be non-zero")]
     InconsistentBootstrapEvidence,
-    #[error("summary operation cost {0} must be finite and non-negative, got {1}")]
+    #[error("required summary operation cost {0} must be finite and positive, got {1}")]
     InvalidOperationCost(&'static str, f64),
     #[error("required analytical input {0} is missing or zero")]
     MissingOrZero(&'static str),

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

@@ -50,7 +50,7 @@ use std::rc::Rc;
 
 use asap_types::post_asap::{
     GroupingStrategy, HydraParams, SketchAlgorithm, SketchParams, SketchQuery, SummaryExpr,
-    SummaryFamilyType, SummaryNode,
+    SummaryFamilyType, SummaryMaintenanceLifecycleGuarantee, SummaryNode,
 };
 use asap_types::pre_asap::agg_intent::AggIntent;
 use asap_types::pre_asap::expr_ir::ColumnRef;
@@ -99,6 +99,15 @@ pub struct CseCandidate<'a> {
 #[derive(Debug, Clone, Copy, PartialEq, PartialOrd)]
 pub struct Cost(pub f64);
 
+/// One physical summary state and the lifecycle selected for that exact DAG
+/// node. Node identity is preserved so whole-DAG models can bind per-state
+/// evidence without relying on traversal order.
+pub struct CostedSummaryDeployment<'a> {
+    pub summary: &'a SummaryNode,
+    pub guarantee: &'a SummaryMaintenanceLifecycleGuarantee,
+    pub selected_cost: Cost,
+}
+
 impl Cost {
     /// The cost of an operation that costs nothing at all.
     pub const ZERO: Cost = Cost(0.0);
@@ -543,6 +552,17 @@ pub trait CostModel {
         SummaryMaintenanceLifecycleCostInputs::default()
     }
 
+    /// Horizon-aware form used by lifecycle planning. Models whose retention
+    /// objective is capacity rather than byte-seconds can normalize their
+    /// rate so the horizon integral equals one peak-capacity charge.
+    fn summary_maintenance_lifecycle_cost_inputs_for_horizon(
+        &self,
+        summary: &SummaryNode,
+        _horizon: Option<Horizon>,
+    ) -> SummaryMaintenanceLifecycleCostInputs {
+        self.summary_maintenance_lifecycle_cost_inputs(summary)
+    }
+
     /// Physical update/merge/delete support for one concrete summary. The
     /// conservative default advertises no long-lived maintenance capability.
     fn summary_maintenance_capabilities(
@@ -552,12 +572,44 @@ pub trait CostModel {
         SummaryMaintenanceCapabilities::default()
     }
 
+    /// Replace the sum of selected per-state lifecycle costs with a complete
+    /// root-DAG cost. The default preserves legacy models. Evidence-strict
+    /// models return `None` when any root operation is unavailable; callers
+    /// must not then reuse the partial per-state sum.
+    fn complete_summary_candidate_cost(
+        &self,
+        _root: &SummaryNode,
+        _target: Option<&QueryExpr>,
+        deployments: &[CostedSummaryDeployment<'_>],
+        _horizon: Option<Horizon>,
+        _expected_reads: Option<f64>,
+    ) -> Option<Cost> {
+        Some(Cost(
+            deployments
+                .iter()
+                .map(|deployment| deployment.selected_cost.0)
+                .sum(),
+        ))
+    }
+
     /// Cost of evaluating `target` directly from its logical/raw inputs once.
     /// When known, lifecycle-aware materialization compares this fallback with
     /// the aggregate cost of the selected summary deployments.
     fn raw_query_recompute_cost(&self, _target: &QueryExpr) -> Option<Cost> {
         None
     }
+
+    /// Complete raw cost over the comparison context. The default preserves
+    /// per-read models; context-aware models override this when raw input
+    /// cardinality changes between evaluations.
+    fn raw_query_recompute_total_cost(
+        &self,
+        target: &QueryExpr,
+        expected_reads: f64,
+    ) -> Option<Cost> {
+        self.raw_query_recompute_cost(target)
+            .map(|per_read| Cost(per_read.0 * expected_reads))
+    }
 }
 
 fn sketch_state(

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

@@ -69,7 +69,10 @@ impl ComparisonScope {
 
     /// Validate this scope and return its effective query evaluation count.
     pub fn validate(&self) -> Result<u64, AnalyticalCostError> {
-        if self.data_arrival != DataArrival::AtRest {
+        if !matches!(
+            self.data_arrival,
+            DataArrival::AtRest | DataArrival::ContinuouslyIngesting
+        ) {
             return Err(AnalyticalCostError::UnsupportedDataArrival(
                 self.data_arrival,
             ));

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

@@ -2298,9 +2298,21 @@ pub struct PlanSpace<Id> {
 #[derive(Default)]
 pub(crate) struct CandidateCostOverrides {
     costs: HashMap<(*const QueryExpr, *const ReplacementSubDAG), Cost>,
+    raw_costs: HashMap<*const QueryExpr, Cost>,
+    /// Targets for which the caller requested an atomic raw-vs-summary
+    /// decision. Other memo groups continue through ordinary CSE selection.
+    finalized_targets: HashSet<*const QueryExpr>,
 }
 
 impl CandidateCostOverrides {
+    pub(crate) fn finalize_target(&mut self, target: &Rc<QueryExpr>) {
+        self.finalized_targets.insert(Rc::as_ptr(target));
+    }
+
+    fn finalizes(&self, target: &Rc<QueryExpr>) -> bool {
+        self.finalized_targets.contains(&Rc::as_ptr(target))
+    }
+
     pub(crate) fn insert(
         &mut self,
         target: &Rc<QueryExpr>,
@@ -2316,6 +2328,14 @@ impl CandidateCostOverrides {
             .get(&(Rc::as_ptr(target), candidate as *const _))
             .copied()
     }
+
+    pub(crate) fn insert_raw(&mut self, target: &Rc<QueryExpr>, cost: Cost) {
+        self.raw_costs.insert(Rc::as_ptr(target), cost);
+    }
+
+    fn raw(&self, target: &Rc<QueryExpr>) -> Option<Cost> {
+        self.raw_costs.get(&Rc::as_ptr(target)).copied()
+    }
 }
 
 impl<Id> PlanSpace<Id> {
@@ -3129,19 +3149,26 @@ impl<Id> PlanSpace<Id> {
             let effective = effective_uses.get(ptr).copied().unwrap_or(0);
             effective_uses.insert(*ptr, effective);
 
-            let lifecycle_choice = candidate_costs.and_then(|costs| {
-                group
-                    .candidates
-                    .iter()
-                    .filter_map(|candidate| {
-                        costs
-                            .get(&group.target, candidate)
-                            .map(|cost| (candidate, cost))
-                    })
-                    .min_by(|(_, a), (_, b)| a.0.total_cmp(&b.0))
-                    .map(|(candidate, _)| candidate)
-            });
-            let chosen = if lifecycle_choice.is_some() {
+            let lifecycle_choice = candidate_costs
+                .filter(|costs| costs.finalizes(&group.target))
+                .map(|costs| {
+                    let summary = group
+                        .candidates
+                        .iter()
+                        .filter_map(|candidate| {
+                            costs
+                                .get(&group.target, candidate)
+                                .map(|cost| (candidate, cost))
+                        })
+                        .min_by(|(_, a), (_, b)| a.0.total_cmp(&b.0));
+                    match (summary, costs.raw(&group.target)) {
+                        (Some((_, summary_cost)), Some(raw)) if raw.0 <= summary_cost.0 => None,
+                        (Some((candidate, _)), _) => Some(candidate),
+                        (None, Some(_)) => None,
+                        (None, None) => None,
+                    }
+                });
+            let chosen = if let Some(lifecycle_choice) = lifecycle_choice {
                 lifecycle_choice
             } else if cost_model.candidate_cost_covers_complete_plan() {
                 let effective_target = TargetSubDAG::with_consumer_count(&group.target, effective);