refactor: move the unified frontends and physical planner to the crate roots

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

@@ -2828,7 +2828,7 @@ mod tests {
                 ..Default::default()
             }),
         };
-        let root = asap_frontend_promql::unified::lower_promql_workload(&workload, 0)
+        let root = asap_frontend_promql::lower_promql_workload(&workload, 0)
             .unwrap()
             .remove(0);
         // Read one node through every reference (as a planner that shares

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

@@ -51,7 +51,7 @@
 //!   here, so it is deliberately left unhandled (see [`rollup_combinator`]).
 //!   In practice this never matters: `Rate`/`Increase` are constructed with
 //!   `Reduction::PerEntity` (per-series, no `by` at all — see
-//!   `AggIntent::is_per_series` and `asap_frontend_promql::unified::promql::reduction_for`),
+//!   `AggIntent::is_per_series` and `asap_frontend_promql::promql::reduction_for`),
 //!   never `Reduction::Reduce`, so they never carry a `by` set for this
 //!   strategy to compare in the first place; `PerEntity` nodes never match
 //!   ([`bindable_grouped_aggregate`] returns `None` for them). `Increase`'s

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

@@ -35,7 +35,7 @@ pub(crate) fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Rc<Operator
             ..Default::default()
         }),
     };
-    asap_frontend_promql::unified::lower_promql_workload(&workload, 0)
+    asap_frontend_promql::lower_promql_workload(&workload, 0)
         .unwrap()
         .pop()
         .unwrap()

crates/asap-physical-operators/README.md

@@ -14,7 +14,7 @@ thread pool. Poll multiple root streams concurrently when they share inputs.
 
 `operators::Operator` implements native batch sources, scalar values,
 projection, filtering, grouped exact aggregation, semi-join, grouped Sort and
-Limit, vector-to-scalar conversion, Union, and summary construction/merge/readout.
+Limit, vector-to-scalar conversion, Union, and summary construction/merge/evaluation.
 Sort followed by Limit implements grouped ranking; no dedicated TopK physical
 operator is needed. Summary construction updates state batch by batch. End of
 input means the supplied query range or ingestion window is complete.
@@ -47,22 +47,21 @@ assert!(matches!(batch.rows()[0][0], Value::Int64(-7)));
 # Ok::<(), asap_physical_operators::dag::Error>(())
 ```
 
-`physical_planner::compile` accepts a logical Post-ASAP DAG (`PostAsapDAG`) and typed input contracts.
+`physical_planner::compile` accepts a Planner `PhysicalASAPDAG` and typed input contracts.
 The resulting candidate is instantiated with deployment readers after selection. It rejects unsupported operations and
 schema mismatches before starting a source. Implement `PhysicalOperator` for a
 deployment source, including asynchronous I/O; computation operators remain in
 the library. The public `planner` export identifies the exact Planner types used
 by the crate. The physical compiler currently supports a subset of those types and
-operations; it does not interpret an unknown node as external fallback.
-`unified_physical_planner::compile` does the same for a `PhysicalASAPDAG`: a
+operations; it does not interpret an unknown node as external fallback: a
 node without a native lowering fails compilation, and the deployment runtime
 must execute it.
 
 Plain values preserve Planner scalar/collection types and nullability. Numeric
 arithmetic uses matching Int64 or Float64 inputs; integer overflow is an error.
 Boolean predicates use three-valued logic. Native summary states currently cover
 exact Sum/Count/Min/Max/Rate/Increase, KLL, DDSketch, HLL and Float64 weighted CMS and CountSketch with candidate heaps. Binding checks family,
-parameters and readout compatibility; source batches also validate state payloads.
+parameters and evaluation compatibility; source batches also validate state payloads.
 Existing accumulator algorithms are reused as kernels behind these operators.
 
 This crate is owned by ASAPPlanner. Its `planner-types` dependency is the local
@@ -81,9 +80,9 @@ See [the design](../../docs/design_docs/physical-planning-and-deployment.md).
 - `operators`: projection, filter, joins, aggregate/window, sort, limit and summary implementations.
 - `sources`: raw-source interface, Scan and the memory connector.
 - `physical_planner`: native operator lowering, typed input contracts and checked instantiation.
-- `summary_kernels`: in-memory summary state over `asap_sketchlib` and exact Planner state: merge, typed readout and update adapters.
-- `readout`: readouts over merged exact summary states.
-- `capability`: explicit kernel, native-batch and typed readout validation.
+- `summary_kernels`: in-memory summary state over `asap_sketchlib` and exact Planner state: merge, typed evaluation and update adapters.
+- `evaluation`: evaluations over merged exact summary states.
+- `capability`: explicit kernel, native-batch and typed evaluation validation.
 
 The `dag`, `factory`, `traits` and `arithmetic` paths are re-exports. They contain no alternative
 execution implementations.
@@ -107,7 +106,7 @@ There is no spill or partitioned parallel execution in this implementation.
 
 ## Physical compilation and deployment inputs
 
-`physical_planner::compile` accepts a Planner `PostAsapDAG`, typed
+`physical_planner::compile` accepts a Planner `PhysicalASAPDAG`, typed
 `InputContract`s and output roots. It returns a reusable `CompiledPhysicalDAG`
 containing selected native operators and no live readers. Compilation validates
 schemas, input ordering, sharing and boundedness before deployment source access.

crates/asap-physical-operators/src/dag/mod.rs

@@ -5,6 +5,4 @@ pub use crate::runtime::{
     Input, Limits, OutputStream, Reservation, RunContext, Scope, SharedValue,
 };
 pub use crate::Error;
-pub use crate::{
-    expressions, operators, unified_physical_planner as planner, unified_sources as scan, values,
-};
+pub use crate::{expressions, operators, physical_planner as planner, sources as scan, values};

crates/asap-physical-operators/src/expressions/mod.rs

@@ -6,15 +6,16 @@ use crate::{
 use planner_types::pre_asap::{ArithmeticOpKind, DataType};
 pub mod arithmetic;
 pub mod binary;
-pub mod unified_planner;
+mod planner;
+pub use planner::CompiledExpression;
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug)]
 pub enum Expression {
     Binary {
         operator: crate::expressions::binary::BinaryOperator,
         left: Box<Expression>,
         right: Box<Expression>,
     },
-    UnifiedPlanner(Box<unified_planner::CompiledExpression>),
+    Planner(Box<CompiledExpression>),
     Column(usize),
     ExactFloat64(usize),
     FiniteFloat64(Box<Expression>),
@@ -53,8 +54,8 @@ pub enum Expression {
     IsNull(Box<Expression>),
 }
 impl Expression {
-    pub fn unified_planner(expression: unified_planner::CompiledExpression) -> Self {
-        Self::UnifiedPlanner(Box::new(expression))
+    pub fn planner(expression: CompiledExpression) -> Self {
+        Self::Planner(Box::new(expression))
     }
     pub(crate) fn dtype(&self, input: &SchemaRef) -> Result<(DataType, bool), Error> {
         use Expression::*;
@@ -100,7 +101,7 @@ impl Expression {
                 };
                 Ok((dtype, n || m))
             }
-            UnifiedPlanner(expression) => {
+            Planner(expression) => {
                 expression.validate_input(input)?;
                 Ok(expression.dtype())
             }
@@ -286,7 +287,7 @@ impl Expression {
                     arithmetic::evaluate_binary(operator, a, b)?
                 }
             }
-            UnifiedPlanner(expression) => expression.evaluate(row)?,
+            Planner(expression) => expression.evaluate(row)?,
             Label { column, name } => {
                 let Value::Map(entries) = &row[*column] else {
                     return Err(invalid("label read requires a map"));
@@ -319,8 +320,7 @@ impl Expression {
                     }
                 }
                 Value::Utf8(
-                    crate::unified_physical_planner::promql_rows::encode_series_identity(&labels)?
-                        .into(),
+                    crate::physical_planner::promql_rows::encode_series_identity(&labels)?.into(),
                 )
             }
             Column(i) => row[*i].clone(),

crates/asap-physical-operators/src/lib.rs

@@ -26,9 +26,8 @@ mod error;
 pub use error::Error;
 pub mod expressions;
 pub mod operators;
+pub mod physical_planner;
 pub mod plan;
 pub mod runtime;
+pub mod sources;
 pub mod values;
-
-pub mod unified_physical_planner;
-pub mod unified_sources;

crates/asap-physical-operators/src/operators/joins/mod.rs

@@ -60,10 +60,8 @@ impl Operator {
         use planner_types::pre_asap::JoinKind;
         let mut joined = left.fields.clone();
         joined.extend(right.fields.clone());
-        let predicate = crate::expressions::unified_planner::CompiledExpression::compile(
-            &predicate.0,
-            &schema(joined.clone()),
-        )?;
+        let predicate =
+            crate::expressions::CompiledExpression::compile(&predicate.0, &schema(joined.clone()))?;
         if predicate.dtype().0 != DataType::Bool {
             return Err(invalid("join predicate must be boolean"));
         }

crates/asap-physical-operators/src/operators/mod.rs

@@ -130,7 +130,7 @@ enum Kind {
     },
     Join {
         kind: planner_types::pre_asap::JoinKind,
-        predicate: Box<crate::expressions::unified_planner::CompiledExpression>,
+        predicate: Box<crate::expressions::CompiledExpression>,
     },
     SummaryBuild {
         family: SummaryFamilyType,

crates/asap-physical-operators/src/unified_physical_planner/candidates.rs → crates/asap-physical-operators/src/physical_planner/candidates.rs

@@ -368,7 +368,7 @@ mod tests {
                 ..Default::default()
             }),
         };
-        let root = asap_frontend_promql::unified::lower_promql_workload(&workload, 0)
+        let root = asap_frontend_promql::lower_promql_workload(&workload, 0)
             .unwrap()
             .remove(0);
         let root = promql_rows::with_series_identity(&root).unwrap();
@@ -401,7 +401,7 @@ mod tests {
     #[test]
     fn candidates_for_all_frontiers_share_one_lowering() {
         let (dag, inputs, root) = grouped_rate();
-        let lowered = || crate::unified_physical_planner::LOWERED_NODES.with(|count| count.get());
+        let lowered = || crate::physical_planner::LOWERED_NODES.with(|count| count.get());
         let before = lowered();
         let compiled = compile(&dag, inputs, &[root]).unwrap();
         let once = lowered() - before;
@@ -455,7 +455,7 @@ mod tests {
         let (retained, _, root) = grouped_rate();
         let ephemeral = with_timing(&retained, |_| QueryTime);
         let inputs = raw_input(&retained);
-        let lowered = || crate::unified_physical_planner::LOWERED_NODES.with(|count| count.get());
+        let lowered = || crate::physical_planner::LOWERED_NODES.with(|count| count.get());
         let before = lowered();
         let compiled = compile(&ephemeral, inputs.clone(), &[root]).unwrap();
         let once = lowered() - before;

crates/asap-physical-operators/src/unified_physical_planner/mod.rs → crates/asap-physical-operators/src/physical_planner/mod.rs

@@ -80,7 +80,7 @@ pub fn bind_with_data_sources<'a>(
     dag: &PhysicalASAPDAG,
     mut sources: BTreeMap<NodeId, Source<'a>>,
     roots: &[NodeId],
-    data_sources: &crate::unified_sources::DataSources,
+    data_sources: &crate::sources::DataSources,
 ) -> Result<PhysicalDAG<'a, Batch, SchemaRef>, Error> {
     let restored = logical::restore(dag)?;
     // Only resolve scans reachable below the selected input boundaries.
@@ -906,8 +906,8 @@ fn expression(
     input: &SchemaRef,
 ) -> Result<Expression, Error> {
     let expr = local_scalar(expr)?;
-    Ok(Expression::unified_planner(
-        crate::expressions::unified_planner::CompiledExpression::compile(&expr, input)?,
+    Ok(Expression::planner(
+        crate::expressions::CompiledExpression::compile(&expr, input)?,
     ))
 }
 

crates/asap-physical-operators/src/unified_sources/mod.rs → crates/asap-physical-operators/src/sources/mod.rs

@@ -1,6 +1,6 @@
 //! Raw data access. Connectors provide rows; Scan owns Planner predicate semantics.
 use crate::{
-    expressions::unified_planner::CompiledExpression,
+    expressions::CompiledExpression,
     plan::PhysicalOperator,
     runtime::{Input, OutputStream, RunContext},
     values::{Batch, SchemaRef, Value},

crates/asap-physical-operators/tests/current_series_heap.rs

@@ -1,13 +1,12 @@
 //! Spatial heap weights come from a fresh instant vector, never sample history.
-#[path = "unified_common/mod.rs"]
 mod common;
 use asap_physical_operators::{
     operators::Operator,
-    runtime::{Limits, RunContext, Scope},
-    unified_physical_planner::{
+    physical_planner::{
         promql_rows::{decode_series_identity, series_row, SERIES_IDENTITY_COLUMN},
         CompiledPhysicalDAG, InputContract, Source,
     },
+    runtime::{Limits, RunContext, Scope},
     values::{Batch, Value},
 };
 use common::compile_physical_asap_dag;
@@ -273,7 +272,7 @@ fn current_series_observes_resource_limits() {
 
 #[test]
 fn identity_encoding_is_lossless_and_rejects_noncanonical_inputs() {
-    use asap_physical_operators::unified_physical_planner::promql_rows::encode_series_identity;
+    use asap_physical_operators::physical_planner::promql_rows::encode_series_identity;
     let labels = BTreeMap::from([
         ("a".into(), "quote\"slash\\".into()),
         ("other".into(), "".into()),
@@ -296,9 +295,7 @@ fn identity_encoding_is_lossless_and_rejects_noncanonical_inputs() {
 // test does not manually assemble the computation or its dependency edges.
 #[test]
 fn planner_current_series_candidate_compiles_with_dynamic_identity() {
-    use asap_physical_operators::unified_physical_planner::{
-        compile, promql_rows::with_series_identity,
-    };
+    use asap_physical_operators::physical_planner::{compile, promql_rows::with_series_identity};
     use planner_types::{types::AccuracyTarget, workload::*};
     use std::rc::Rc;
     let workload = PlanningWorkload {
@@ -325,7 +322,7 @@ fn planner_current_series_candidate_compiles_with_dynamic_identity() {
             ..Default::default()
         }),
     };
-    let original = asap_frontend_promql::unified::lower_promql_workload(&workload, 0)
+    let original = asap_frontend_promql::lower_promql_workload(&workload, 0)
         .unwrap()
         .remove(0);
     let open_root = Rc::new(original.clone());
@@ -336,7 +333,7 @@ fn planner_current_series_candidate_compiles_with_dynamic_identity() {
         .candidate(&open_root)
         .unwrap();
     let snapshot_program =
-        asap_physical_operators::unified_physical_planner::promql_rows::compile_current_series_evaluation(
+        asap_physical_operators::physical_planner::promql_rows::compile_current_series_evaluation(
             &open_selected,
         )
         .unwrap();

crates/asap-physical-operators/tests/deployment_computation.rs

@@ -1,11 +1,10 @@
 //! Planner-selected PromQL computation compiles from the timed DAG alone;
 //! the deployment supplies only raw rows at the ingestion frontier.
-#[path = "unified_common/mod.rs"]
 mod common;
 use asap_physical_operators::{
     operators::Operator,
+    physical_planner::{compile, promql_rows, CompiledPhysicalDAG, InputContract, Source},
     runtime::{Limits, RunContext, Scope},
-    unified_physical_planner::{compile, promql_rows, CompiledPhysicalDAG, InputContract, Source},
     values::{Batch, Value},
 };
 use common::compile_physical_asap_dag;
@@ -44,7 +43,7 @@ fn lower_with(query: &str, accuracy: AccuracyTarget) -> Rc<planner_types::ir::Op
             ..Default::default()
         }),
     };
-    asap_frontend_promql::unified::lower_promql_workload(&workload, 0)
+    asap_frontend_promql::lower_promql_workload(&workload, 0)
         .unwrap()
         .remove(0)
 }

crates/asap-physical-operators/tests/unified_node_ids.rs → crates/asap-physical-operators/tests/node_ids.rs

@@ -1,7 +1,6 @@
 //! Compilation rejects Planner node IDs outside the range its helper IDs avoid.
-#[path = "unified_common/mod.rs"]
 mod common;
-use asap_physical_operators::unified_physical_planner::compile;
+use asap_physical_operators::physical_planner::compile;
 use common::compile_physical_asap_dag;
 use planner_types::ir::{NonASAPOp, Operator, OperatorNode, Predicate, ScalarExpr};
 use planner_types::pre_asap::{DataType, Field, ScalarValue, Schema};

crates/asap-physical-operators/tests/physical_dag.rs

@@ -877,7 +877,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_physical_operators::dag::expressions::unified_planner::CompiledExpression;
+    use asap_physical_operators::dag::expressions::CompiledExpression;
     use planner_types::pre_asap::{CompareOpKind, ScalarValue};
 
     let input_schema = schema(&[(
@@ -900,9 +900,7 @@ fn planner_expressions_preserve_collection_and_nullable_types() {
         input_schema.clone(),
         vec![(
             "count".into(),
-            Expression::unified_planner(
-                CompiledExpression::compile(&access, &input_schema).unwrap(),
-            ),
+            Expression::planner(CompiledExpression::compile(&access, &input_schema).unwrap()),
         )],
     )
     .unwrap();
@@ -939,9 +937,7 @@ fn planner_expressions_preserve_collection_and_nullable_types() {
         vec![1],
         Operator::filter(
             projected.clone(),
-            Expression::unified_planner(
-                CompiledExpression::compile(&predicate, &projected).unwrap(),
-            ),
+            Expression::planner(CompiledExpression::compile(&predicate, &projected).unwrap()),
         )
         .unwrap(),
     )
@@ -1162,7 +1158,7 @@ fn assert_weighted_rate_topk(count_sketch: bool) {
 // The grouped temporal reducer's sample schema must survive physical Sort/Limit binding.
 #[test]
 fn grouped_temporal_schema_compiles_and_executes_topk() {
-    use asap_physical_operators::unified_physical_planner::{
+    use asap_physical_operators::physical_planner::{
         compile_node, CompiledPhysicalDAG, InputContract, Source,
     };
     use planner_types::ir::physical_export::{PhysicalASAPDAGNode, PhysicalASAPOperatorPayload};
@@ -1273,7 +1269,7 @@ fn grouped_temporal_schema_compiles_and_executes_topk() {
 // A certified candidate set must have authoritative values for every key, including after recovery.
 #[test]
 fn certified_pruning_rejects_missing_authoritative_values_after_recovery() {
-    use asap_physical_operators::unified_physical_planner::{
+    use asap_physical_operators::physical_planner::{
         compile_node, CompiledPhysicalDAG, InputContract, Source,
     };
     use planner_types::{
@@ -1385,7 +1381,7 @@ fn certified_pruning_rejects_missing_authoritative_values_after_recovery() {
 // Precompute arithmetic must match population/window identities, never zip arrival order.
 #[test]
 fn compiled_ingestion_binary_preserves_alignment_and_rejects_missing_updates() {
-    use asap_physical_operators::unified_physical_planner::{
+    use asap_physical_operators::physical_planner::{
         compile_node, CompiledPhysicalDAG, InputContract, Source,
     };
     use planner_types::{

crates/asap-physical-operators/tests/physical_plan_recovery.rs

@@ -2,7 +2,7 @@
 //! Deployments choose the encoding; JSON is used here only as a test format.
 use asap_physical_operators::{
     operators::{Operator, SortKey},
-    unified_physical_planner::{CompiledPhysicalDAG, InputContract},
+    physical_planner::{CompiledPhysicalDAG, InputContract},
 };
 use planner_types::{
     post_asap::{Field, FieldDataType},
@@ -74,7 +74,7 @@ fn recovery_retains_selected_operator_and_rejects_invalid_contracts() {
 
 #[test]
 fn candidate_recovery_preserves_materialization_boundary() {
-    use asap_physical_operators::unified_physical_planner::CompiledPhysicalPlan;
+    use asap_physical_operators::physical_planner::CompiledPhysicalPlan;
     let precompute = sorted();
     let output = InputContract::bounded(precompute.output_contract(1).unwrap().schema);
     let query = CompiledPhysicalDAG::from_operators(