Merge pull request #336 from ProjectASAP/feat/cost-streaming-planner-integration

feat(cost): rank complete streaming deployments

Author: zzylol

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

@@ -108,6 +108,16 @@ pub struct CostedSummaryDeployment<'a> {
     pub selected_cost: Cost,
 }
 
+/// Complete physical-plan result for one lifecycle combination. Models that
+/// enumerate multiple deployment implementations can expose the stable
+/// provider-owned identity of the winning alternative without adding a
+/// closed window-framework enum to the planner.
+#[derive(Debug, Clone, PartialEq)]
+pub struct CompleteSummaryCandidateEstimate {
+    pub cost: Cost,
+    pub physical_plan_id: Option<String>,
+}
+
 impl Cost {
     /// The cost of an operation that costs nothing at all.
     pub const ZERO: Cost = Cost(0.0);
@@ -592,6 +602,23 @@ pub trait CostModel {
         ))
     }
 
+    /// Complete-plan estimate plus the identity of the selected physical
+    /// implementation, when the model searched more than one implementation.
+    fn complete_summary_candidate_estimate(
+        &self,
+        root: &SummaryNode,
+        target: Option<&QueryExpr>,
+        deployments: &[CostedSummaryDeployment<'_>],
+        horizon: Option<Horizon>,
+        expected_reads: Option<f64>,
+    ) -> Option<CompleteSummaryCandidateEstimate> {
+        self.complete_summary_candidate_cost(root, target, deployments, horizon, expected_reads)
+            .map(|cost| CompleteSummaryCandidateEstimate {
+                cost,
+                physical_plan_id: None,
+            })
+    }
+
     /// 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.

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

@@ -205,7 +205,7 @@ pub use accuracy::{
     PropagationStats, WorkloadAccuracyEvidence,
 };
 pub use accuracy_reconciliation::AccuracyReconciliationStrategy;
-pub use cost_model::{CostModel, DefaultCostModel};
+pub use cost_model::{CompleteSummaryCandidateEstimate, CostModel, DefaultCostModel};
 pub use explanation::{
     explain_replacements, explain_replacements_with, ExplanationKind, ReplacementExplanation,
 };

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

@@ -329,3 +329,12 @@ pub struct StreamingRawInputEvidence {
     pub ingestion_rate_per_second: f64,
     pub physical_dag: EvidenceBackedPhysicalDag,
 }
+
+/// One complete provider-enumerated physical implementation of the selected
+/// streaming summary DAG. The identifier is stable provenance; concrete
+/// framework selection is performed by ranking these complete alternatives.
+#[derive(Debug, Clone)]
+pub struct StreamingPhysicalPlanAlternative {
+    pub physical_plan_id: String,
+    pub node_evidence: StreamingNodeEvidence,
+}

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

@@ -9,7 +9,7 @@ use std::rc::Rc;
 
 use asap_types::post_asap::{
     SketchAlgorithm, SummaryExpr, SummaryMaintenanceLifecycle,
-    SummaryMaintenanceLifecycleGuarantee, SummaryMaintenanceMode, SummaryNode,
+    SummaryMaintenanceLifecycleGuarantee, SummaryNode,
 };
 use asap_types::pre_asap::{
     agg_intent::AggIntent, CompareOpKind, InfoMatcher, Predicate, QueryExpr, Source,
@@ -18,14 +18,15 @@ use asap_types::workload::{DataArrival, DataWorkload, QueryRecurrence, RepeatedD
 use serde::{Deserialize, Serialize};
 
 use crate::analytical_cost::ExecutionMultiplicity;
+#[cfg(test)]
+use crate::analytical_cost::PhysicalNodeEvidence;
 use crate::analytical_cost::{
-    estimate_physical_dag, AnalyticalCostError, EvidenceBackedPhysicalDag, PhysicalOperator,
-    ResourceCalibration, ResourceEstimate,
+    estimate_physical_dag, AnalyticalCostError, EvidenceBackedPhysicalDag, PhysicalDagNode,
+    PhysicalOperator, ResourceCalibration, ResourceEstimate,
+};
+use crate::cost_model::{
+    CompleteSummaryCandidateEstimate, Cost, CostModel, CostedSummaryDeployment, DefaultCostModel,
 };
-#[cfg(test)]
-use crate::analytical_cost::{PhysicalDagNode, PhysicalNodeEvidence};
-use crate::cost_model::CostedSummaryDeployment;
-use crate::cost_model::{Cost, CostModel, DefaultCostModel};
 #[cfg(test)]
 use crate::physical_operator_statistics::UnaryEdgeStatistics;
 use crate::physical_operator_statistics::{ComparisonScope, EdgeStatistics, OperatorStatistics};