refactor(plan-selection): delete the unused physical-DAG, storage and handoff cost estimators

crates/executor/README.md

@@ -45,7 +45,7 @@ let run = RunContext::new(
 let mut output = plan.execute(&[1], run)?.remove(0);
 let batch = block_on(output.next()).unwrap()?;
 assert!(matches!(batch.to_rows()[0][0], Value::Int64(-7)));
-# Ok::<(), asap_executor::dag::Error>(())
+# Ok::<(), asap_executor::Error>(())
 ```
 
 `physical_planner::compile` accepts a Planner `PhysicalASAPDAG` and typed input contracts.

crates/executor/src/lib.rs

@@ -27,8 +27,6 @@ pub use planner_types as planner;
 /// columns ([`values::Batch::try_from_columns`], [`values::Batch::column`]).
 pub use arrow;
 
-pub mod dag;
-
 pub mod evaluation;
 
 mod error;

crates/executor/tests/blocking_resources.rs

@@ -238,7 +238,7 @@ fn weighted_summary_build_yields_within_a_batch() {
         input,
         family,
         vec![0],
-        asap_executor::dag::operators::RowWeight::Column(1),
+        asap_executor::operators::RowWeight::Column(1),
         vec![],
     )
     .unwrap();

crates/executor/tests/current_series_heap.rs

@@ -175,7 +175,7 @@ fn spatial_heap_ranks_latest_values_in_independent_runs() {
             schema(),
             family,
             vec![3],
-            asap_executor::dag::operators::RowWeight::Column(1),
+            asap_executor::operators::RowWeight::Column(1),
             vec![2],
         )
         .unwrap();

crates/executor/tests/hydra_cms_execution.rs

@@ -1,12 +1,10 @@
 //! A Hydra-grouped Count-Min `SummaryAgg` compiles through the physical
 //! planner and answers each group's item point counts from one shared grid.
 use asap_executor::{
-    dag::{
-        operators::{Operator, RowWeight},
-        values::{Batch, SchemaRef, Value},
-        Limits, RunContext, Scope,
-    },
+    operators::{Operator, RowWeight},
     physical_planner::{compile_node, CompiledPhysicalDAG, InputContract, Source},
+    runtime::{Limits, RunContext, Scope},
+    values::{Batch, SchemaRef, Value},
 };
 use futures::{executor::block_on, StreamExt};
 use planner_types::ir::operator::{GroupKeys, Reduction};

crates/executor/tests/keyed_summary_row_weight.rs

@@ -2,12 +2,11 @@
 //! group) adds each row's `RowWeight` to the row's item: the value of a
 //! weight column, or a constant such as 1 for a row count.
 use asap_executor::{
-    dag::{
-        operators::{Operator, RowWeight},
-        values::{Batch, SchemaRef, Value},
-        Limits, PhysicalDAG, RunContext, Scope,
-    },
+    operators::{Operator, RowWeight},
     physical_planner::compile_node,
+    plan::PhysicalDAG,
+    runtime::{Limits, RunContext, Scope},
+    values::{Batch, SchemaRef, Value},
 };
 use futures::{executor::block_on, StreamExt};
 use planner_types::ir::operator::{GroupKeys, Reduction};

crates/executor/tests/physical_dag.rs

@@ -1,10 +1,9 @@
 //! Acceptance tests use the library directly, without either backend engine.
 use asap_executor::{
-    dag::{
-        operators::{Expression, Operator, Reduction, SortKey},
-        values::{Batch, SchemaRef, Value},
-        Limits, PhysicalDAG, RunContext, Scope,
-    },
+    operators::{Expression, Operator, Reduction, SortKey},
+    plan::PhysicalDAG,
+    runtime::{Limits, RunContext, Scope},
+    values::{Batch, SchemaRef, Value},
     Statistic,
 };
 use futures::{executor::block_on, StreamExt};
@@ -498,7 +497,7 @@ fn exact_state_and_family_validation() {
 // Planner binding rejects unknown computation instead of accepting a fallback.
 #[test]
 fn bind_physical_asap_dag_before_execution() {
-    use asap_executor::dag::planner::bind;
+    use asap_executor::physical_planner::bind;
     use planner_types::ir::properties::ExecutionDataState;
     use planner_types::ir::scalar::{ArithmeticOpKind, ScalarValue};
     use std::collections::BTreeMap;
@@ -553,7 +552,7 @@ fn bind_physical_asap_dag_before_execution() {
         roots: vec![1],
         scalar_roots: vec![],
     };
-    let sources = || -> BTreeMap<u64, asap_executor::dag::planner::Source<'static>> {
+    let sources = || -> BTreeMap<u64, asap_executor::physical_planner::Source<'static>> {
         BTreeMap::from([(
             0,
             Box::new(
@@ -562,7 +561,7 @@ fn bind_physical_asap_dag_before_execution() {
                     vec![Batch::try_new(schema.clone(), vec![vec![Value::Float64(1.)]]).unwrap()],
                 )
                 .unwrap(),
-            ) as asap_executor::dag::planner::Source<'static>,
+            ) as asap_executor::physical_planner::Source<'static>,
         )])
     };
     let native = bind(&dag, sources(), &[1]).unwrap();
@@ -612,7 +611,7 @@ fn empty_exact_count_is_an_integer_state_evaluation() {
 // A deployment source cannot pass a different row shape to bound expressions.
 #[test]
 fn source_batches_must_match_the_bound_schema() {
-    use asap_executor::dag::{self, PhysicalOperator};
+    use asap_executor::{physical_planner, plan::PhysicalOperator, runtime};
     use planner_types::ir::properties::ExecutionDataState;
     use std::{cell::Cell, collections::BTreeMap, rc::Rc};
     struct WrongSource {
@@ -634,9 +633,9 @@ fn source_batches_must_match_the_bound_schema() {
         }
         fn start<'a>(
             &'a self,
-            _: Vec<dag::Input<'a, Batch>>,
+            _: Vec<runtime::Input<'a, Batch>>,
             _: RunContext,
-        ) -> Result<dag::OutputStream<'a, Batch>, dag::Error> {
+        ) -> Result<runtime::OutputStream<'a, Batch>, asap_executor::Error> {
             self.starts.set(self.starts.get() + 1);
             Ok(
                 futures::stream::once(async { Batch::try_new(schema(&[]), vec![vec![]]) })
@@ -667,14 +666,14 @@ fn source_batches_must_match_the_bound_schema() {
     let source = Box::new(WrongSource {
         schema: expected,
         starts: starts.clone(),
-    }) as dag::planner::Source<'static>;
-    let native = dag::planner::bind(&plan, BTreeMap::from([(0, source)]), &[0]).unwrap();
+    }) as physical_planner::Source<'static>;
+    let native = physical_planner::bind(&plan, BTreeMap::from([(0, source)]), &[0]).unwrap();
     assert_eq!(starts.get(), 0);
     let context = RunContext::new(query(), Limits::default()).unwrap();
     let mut output = native.execute(&[0], context).unwrap().remove(0);
     assert!(matches!(
         block_on(output.next()),
-        Some(Err(dag::Error::AtNode { node: 0, .. }))
+        Some(Err(asap_executor::Error::AtNode { node: 0, .. }))
     ));
     assert_eq!(starts.get(), 1);
 }
@@ -720,7 +719,7 @@ fn extrema_preserve_numeric_values_in_the_presence_of_nan() {
 // Planner wire nodes, including grouping and edge roles, are executable at either phase.
 #[test]
 fn planner_semijoin_sort_limit_contract_at_both_phases() {
-    use asap_executor::dag::planner::{bind, Source};
+    use asap_executor::physical_planner::{bind, Source};
     use planner_types::ir::properties::*;
     use planner_types::ir::schema::*;
 
@@ -888,7 +887,7 @@ fn planner_semijoin_sort_limit_contract_at_both_phases() {
 // Planner scalar signatures, collection access and null predicates share native execution.
 #[test]
 fn planner_expressions_preserve_collection_and_nullable_types() {
-    use asap_executor::dag::expressions::CompiledExpression;
+    use asap_executor::expressions::CompiledExpression;
     use planner_types::ir::scalar::{CompareOpKind, ScalarValue};
 
     let input_schema = schema(&[(
@@ -1114,7 +1113,7 @@ fn assert_weighted_rate_topk(count_sketch: bool) {
         rates.schema(),
         family,
         vec![0],
-        asap_executor::dag::operators::RowWeight::Column(3),
+        asap_executor::operators::RowWeight::Column(3),
         vec![1],
     )
     .unwrap();

crates/executor/tests/raw_scan.rs

@@ -1,9 +1,10 @@
 //! Scan acceptance uses the public connector contract and Planner physical DAGs.
-use asap_executor::dag::{
-    planner::bind_with_data_sources,
-    scan::{DataSources, MemorySource, RawSource},
+use asap_executor::{
+    physical_planner::bind_with_data_sources,
+    runtime::{Limits, OutputStream, RunContext, Scope},
+    sources::{DataSources, MemorySource, RawSource},
     values::{Batch, SchemaRef, Value},
-    Error, Limits, OutputStream, RunContext, Scope,
+    Error,
 };
 use futures::{executor::block_on, stream, StreamExt};
 use planner_types::ir::operator::{GroupKeys, Source};

crates/executor/tests/weighted_topk_binding.rs

@@ -1,10 +1,10 @@
 //! Planner output binds directly to the shared runtime at a declared rate-value frontier.
 mod common;
-use asap_executor::dag::{
+use asap_executor::{
     operators::Operator,
-    planner::{compile, InputContract, Source},
+    physical_planner::{compile, InputContract, Source},
+    runtime::{Limits, RunContext, Scope},
     values::{Batch, Value},
-    Limits, RunContext, Scope,
 };
 use common::{compile_physical_asap_dag, logical_candidates};
 use futures::{executor::block_on, StreamExt};

crates/integration-tests/tests/physical_common/mod.rs

@@ -84,9 +84,9 @@ pub fn execute_raw_rows(
     root: &std::rc::Rc<asap_types::ir::OperatorNode>,
     rows: Vec<Vec<asap_executor::values::Value>>,
 ) -> Vec<Vec<asap_executor::values::Value>> {
-    use asap_executor::dag::{
-        planner::bind_with_data_sources,
-        scan::{DataSources, MemorySource},
+    use asap_executor::{
+        physical_planner::bind_with_data_sources,
+        sources::{DataSources, MemorySource},
     };
     use asap_types::ir::physical_export::PhysicalASAPOperatorPayload;
     use asap_types::ir::NonASAPOp;