refactor(planner): port the extracted Stage 1 crate to derived coverage and flat DAGs

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW

Author: zzylol

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

@@ -47,11 +47,6 @@ pub(crate) fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Rc<Operator
 // `OperatorNode::new_shared`, so a fixture is exactly what a front end
 // would hand the planner.
 
-use asap_types::ir::operator::AggIntent;
-use asap_types::ir::operator::{GroupKeys, Reduction, Source};
-use asap_types::ir::schema::{ColumnId, DataType, Field, Schema};
-use asap_types::ir::{apply_materialization_timings, MaterializationAssignment, TimingMemo};
-use asap_types::ir::{NonASAPOp, Predicate, ScalarExpr, TimeRangeKind};
 use asap_types::ir::operator::agg_intent::AggIntent;
 use asap_types::ir::operator::operator_properties::{Reduction, Source};
 use asap_types::ir::schema::{ColumnId, DataType, Field, Schema};
@@ -127,86 +122,3 @@ pub(crate) fn agg(
 ) -> Rc<OperatorNode> {
     aggregate(Reduction::by(by), vec![intent], vec![], None, child)
 }
-
-/// `intent without (excluded)`.
-pub(crate) fn without_agg(
-    excluded: Vec<ColumnId>,
-    intent: AggIntent,
-    child: Rc<OperatorNode>,
-) -> Rc<OperatorNode> {
-    aggregate(
-        Reduction::Reduce(GroupKeys::without(excluded)),
-        vec![intent],
-        vec![],
-        None,
-        child,
-    )
-}
-
-/// A per-entity (per-series) single-measure aggregate.
-pub(crate) fn agg_per_entity(intent: AggIntent, child: Rc<OperatorNode>) -> Rc<OperatorNode> {
-    aggregate(Reduction::PerEntity, vec![intent], vec![], None, child)
-}
-
-pub(crate) fn filter(pred: ScalarExpr, child: Rc<OperatorNode>) -> Rc<OperatorNode> {
-    OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Filter {
-        pred: Predicate(pred),
-        child,
-    }))
-    .unwrap()
-}
-
-pub(crate) fn dedup(cols: Vec<ColumnId>, child: Rc<OperatorNode>) -> Rc<OperatorNode> {
-    OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Dedup {
-        cols,
-        child,
-    }))
-    .unwrap()
-}
-
-/// An explicit range selector `child[range]`.
-pub(crate) fn time_range(range: Duration, child: Rc<OperatorNode>) -> Rc<OperatorNode> {
-    OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::TimeRange {
-        range,
-        kind: TimeRangeKind::Range,
-        child,
-    }))
-    .unwrap()
-}
-
-/// `root` timed under the default (every summary at query time)
-/// assignment — the shape export and the post-ASAP validators consume.
-pub(crate) fn timed(root: &Rc<OperatorNode>) -> Rc<OperatorNode> {
-    apply_materialization_timings(
-        root,
-        &MaterializationAssignment::all_query_time(),
-        &mut TimingMemo::new(),
-    )
-    .expect("default materialization timings apply")
-}
-
-/// `root` timed with every summary maintained at ingestion time.
-pub(crate) fn maintained(root: &Rc<OperatorNode>) -> Rc<OperatorNode> {
-    apply_materialization_timings(
-        root,
-        &MaterializationAssignment::all_ingestion_time(),
-        &mut TimingMemo::new(),
-    )
-    .expect("maintained materialization timings apply")
-}
-
-/// Time `root` under the default materialization assignment (which runs every
-/// data-state / population-contract check) and export it as a physical ASAP DAG.
-pub(crate) fn time_and_export(
-    root: &Rc<OperatorNode>,
-) -> Result<
-    asap_types::ir::physical_export::PhysicalASAPDAG,
-    asap_types::ir::properties::ExecutionDataStateError,
-> {
-    let timed = apply_materialization_timings(
-        root,
-        &MaterializationAssignment::all_query_time(),
-        &mut TimingMemo::new(),
-    )?;
-    asap_types::ir::physical_export::compile_physical_asap_dag(&timed)
-}

crates/devtools/src/bin/stage_pipeline.rs

@@ -30,12 +30,8 @@ use asap_aware_mapping::PlanningModels;
 use asap_logical_optimizer::pass1::logical_candidates::{
     choice_index, enumerate_local_logical_candidates, LocalLogicalCandidates,
 };
-use asap_aware_mapping::plan_selection::{select_exhaustive, Selection, MAX_ENUMERATED_CANDIDATES};
-use asap_aware_mapping::{PlanningModels, Realization};
-use asap_types::ir::flat::{flatten, FlatDag};
 use asap_logical_optimizer::Realization;
-use asap_types::ir::physical_export::{
-    compile_logical_asap_workload, LogicalASAPDAG, LogicalASAPDAGDocument};
+use asap_types::ir::flat::{flatten, FlatDag};
 use asap_types::ir::schema::SketchAlgorithm;
 use asap_types::ir::schema_support::with_promql_series_identity;
 use asap_types::ir::{OperatorNode, QueryRoot};

crates/frontend-sql/tests/maintained_population.rs

@@ -1,16 +1,16 @@
 //! SQL and PromQL use the same shared-state rule without sharing membership semantics.
 use asap_frontend_sql::{lower_sql, SqlCatalog};
 use asap_logical_optimizer::pass1::maintained_population::MaintainedPopulationStrategy;
-use asap_types::ir::cse::share_common_sub_dags;
-use asap_types::ir::physical_export::compile_physical_asap_dag;
 use asap_types::ir::operator::maintained_population::{MaintainedPopulation, PopulationInput};
 use asap_types::ir::schema::{DataType, Field, Schema};
 use asap_types::{
     ir::{
         apply_materialization_timings, cse::share_common_sub_dags,
         physical_export::compile_physical_asap_dag, ASAPOp, MaterializationAssignment, NonASAPOp,
-        Operator, OperatorNode, TimingMemo},
-    types::AccuracyTarget};
+        Operator, OperatorNode, TimingMemo,
+    },
+    types::AccuracyTarget,
+};
 use std::rc::Rc;
 
 async fn aggregate(q: &str) -> Rc<OperatorNode> {

crates/integration-tests/tests/promql_to_post_asap.rs

@@ -14,8 +14,6 @@ use asap_integration_tests::fixtures::lower_promql;
 use asap_integration_tests::post_asap::{
     maintained, maintained_post_asap_dag, post_asap_dag, timed,
 };
-use asap_types::ir::operator::Reduction;
-use asap_types::ir::physical_export::PhysicalASAPOperatorPayload;
 use asap_logical_optimizer::accuracy::{
     AccuracyEvidenceProvider, DefaultAccuracyModel, EqualSplitAllocator, PropagationStats,
     QuantileInputDomain,
@@ -27,8 +25,8 @@ use asap_logical_optimizer::{
     search_workload, search_workload_with_targets, ASAPStrategies, AccuracyModel, Replacement,
     ReplacementStrategy, ReplacementSubDAG, TargetSubDAG,
 };
-use asap_types::ir::physical_export::{ PhysicalASAPOperatorPayload};
-use asap_types::ir::operator::operator_properties::Reduction;
+use asap_types::ir::operator::Reduction;
+use asap_types::ir::physical_export::PhysicalASAPOperatorPayload;
 use asap_types::ir::properties::CompositionOperator;
 use asap_types::ir::scalar::ColumnRef;
 use asap_types::ir::schema::DataType;

crates/integration-tests/tests/sql_to_post_asap.rs

@@ -23,16 +23,13 @@ use asap_aware_mapping::plan_selection::candidate_selection::global_selection;
 use asap_aware_mapping::DefaultCostModel;
 use asap_frontend_sql::{lower_sql, lower_sql_dialect, SqlCatalog};
 use asap_integration_tests::post_asap::post_asap_dag;
-use asap_types::ir::operator::Reduction;
-use asap_types::ir::physical_export::{EdgeRole, PhysicalASAPNodeId, PhysicalASAPOperatorPayload};
 use asap_logical_optimizer::pass1::replacement::{retain_exact, RealizationError};
 use asap_logical_optimizer::{
     search_workload, ASAPStrategies, Replacement, ReplacementStrategy, ReplacementSubDAG,
     TargetSubDAG,
 };
-use asap_types::ir::physical_export::{
-    EdgeRole, PhysicalASAPNodeId, PhysicalASAPOperatorPayload};
-use asap_types::ir::operator::operator_properties::Reduction;
+use asap_types::ir::operator::Reduction;
+use asap_types::ir::physical_export::{EdgeRole, PhysicalASAPNodeId, PhysicalASAPOperatorPayload};
 use asap_types::ir::scalar::ColumnRef;
 use asap_types::ir::schema::{DataType, Field, Schema};
 use asap_types::ir::schema::{

crates/logical-optimizer/src/pass1/replacement.rs

@@ -351,7 +351,6 @@ use std::collections::{HashMap, HashSet, VecDeque};
 use asap_types::ir::cse::{share_common_sub_dags, structural_hash, HashCache};
 use asap_types::ir::operator::agg_intent::{agg_is_mergeable, AggIntent};
 use asap_types::ir::operator::operator_properties::{BinaryOpKind, JoinKind, Reduction};
-use asap_types::ir::properties::summary_coverage::{CoverageRegion, SummaryCoverage};
 use asap_types::ir::properties::timing::validate_maintained;
 use asap_types::ir::properties::{
     AccuracyError, CompositionOperator, GuaranteeSource, ResultGuarantee,
@@ -4784,10 +4783,8 @@ fn relink_agg_child(node: &Rc<OperatorNode>, new_child: &Rc<OperatorNode>) -> Rc
             if Rc::ptr_eq(child, new_child) {
                 return Rc::clone(node);
             }
-            // The same summary over a re-placed input keeps its coverage.
-            let rebuilt = std::rc::Rc::new(OperatorNode {
-                coverage: node.coverage.clone(),
-                ..OperatorNode::with_schema(
+            let rebuilt = std::rc::Rc::new(
+                OperatorNode::with_schema(
                     asap_types::ir::Operator::ASAP(ASAPOp::SummaryAgg {
                         child: Rc::clone(new_child),
                         family: family.clone(),
@@ -4798,8 +4795,8 @@ fn relink_agg_child(node: &Rc<OperatorNode>, new_child: &Rc<OperatorNode>) -> Rc
                     }),
                     node.schema.clone(),
                 )
-                .with_guarantee(node.guarantee.clone())
-            });
+                .with_guarantee(node.guarantee.clone()),
+            );
             match validate_maintained(&rebuilt, ExecutionTiming::IngestionTime) {
                 Ok(_) => rebuilt,
                 Err(_) => Rc::clone(node),
@@ -8400,20 +8397,13 @@ mod tests {
         }
     }
 
-    // Whole-source coverage names one source; over two it is not declared.
-    #[test]
-    fn whole_source_coverage_needs_exactly_one_source() {
-        let left = metric_scan(&["job"]);
-        let right = crate::test_support::scan("n", left.schema.clone());
-        assert!(whole_source_coverage(&left).is_some());
-        let join = OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Join {
-            kind: asap_types::ir::operator::operator_properties::JoinKind::Inner,
-            pred: equi_pred(0, 2),
-            left,
-            right,
-        }))
-        .unwrap();
-        assert_eq!(whole_source_coverage(&join), None);
+    fn equi_pred(left: ColumnId, right: ColumnId) -> Predicate {
+        Predicate(ScalarExpr::Compare {
+            left: Box::new(ScalarExpr::Column(left)),
+            op: asap_types::ir::scalar::CompareOpKind::Eq,
+            right: Box::new(ScalarExpr::Column(right)),
+            semantics: asap_types::ir::ExprSemantics::Sql,
+        })
     }
 
     #[test]

crates/logical-optimizer/src/test_support.rs

@@ -50,12 +50,10 @@ pub(crate) fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Rc<Operator
 
 use std::time::Duration;
 
-use asap_types::ir::operator::agg_intent::AggIntent;
-use asap_types::ir::operator::operator_properties::{GroupKeys, Reduction, Source};
-use asap_types::ir::properties::timing::{
-    apply_materialization_timings, MaterializationAssignment, TimingMemo,
-};
+use asap_types::ir::operator::AggIntent;
+use asap_types::ir::operator::{GroupKeys, Reduction, Source};
 use asap_types::ir::schema::{ColumnId, DataType, Field, Schema};
+use asap_types::ir::{apply_materialization_timings, MaterializationAssignment, TimingMemo};
 use asap_types::ir::{NonASAPOp, Predicate, ScalarExpr, TimeRangeKind};
 
 /// A `TimeSeries("m")` scan over `[ts(0), value(1), labels...]`, time index 0,
@@ -201,13 +199,13 @@ pub(crate) fn maintained(root: &Rc<OperatorNode>) -> Rc<OperatorNode> {
 pub(crate) fn time_and_export(
     root: &Rc<OperatorNode>,
 ) -> Result<
-    asap_types::ir::export::PhysicalASAPDAG,
-    asap_types::ir::properties::execution::ExecutionDataStateError,
+    asap_types::ir::physical_export::PhysicalASAPDAG,
+    asap_types::ir::properties::ExecutionDataStateError,
 > {
     let timed = apply_materialization_timings(
         root,
         &MaterializationAssignment::all_query_time(),
         &mut TimingMemo::new(),
     )?;
-    asap_types::ir::export::compile_physical_asap_dag(&timed)
+    asap_types::ir::physical_export::compile_physical_asap_dag(&timed)
 }