refactor(planner): extract the Stage 2 physical-optimizer crate

Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "crates/sql-function-catalog",
     "crates/asap-aware-mapping",
     "crates/logical-optimizer",
+    "crates/physical-optimizer",
     "crates/frontend-promql",
     "crates/frontend-metricsql",
     "crates/metricsql-common-parser-support",

crates/asap-aware-mapping/Cargo.toml

@@ -3,11 +3,13 @@ name = "asap-aware-mapping"
 version = "0.1.0"
 edition = "2021"
 
-# #509 Stage 2, Stage 3 and the planner facade over the Stage 1 candidates of
-# asap-logical-optimizer. Never depends on a front end.
+# #509 Stage 3 and the planner facade over the Stage 1 candidates of
+# asap-logical-optimizer and the Stage 2 candidates of asap-physical-optimizer.
+# Never depends on a front end.
 [dependencies]
 asap-types = { path = "../types" }
 asap-logical-optimizer = { path = "../logical-optimizer" }
+asap-physical-optimizer = { path = "../physical-optimizer" }
 thiserror = "2"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

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

@@ -1,8 +1,9 @@
-//! `asap-aware-mapping` — #509 Stage 2, Stage 3 and the planner facade, over
-//! the Stage 1 candidates of [`asap_logical_optimizer`].
+//! `asap-aware-mapping` — #509 Stage 3 and the planner facade, over the
+//! Stage 1 candidates of [`asap_logical_optimizer`] and the Stage 2
+//! candidates of [`asap_physical_optimizer`].
 //!
-//! It is being split into one crate per stage (#572): Stage 1 already lives in
-//! `asap-logical-optimizer`; Stage 2 (`physical_candidates`), Stage 3
+//! It is being split into one crate per stage (#572): Stage 1 lives in
+//! `asap-logical-optimizer` and Stage 2 in `asap-physical-optimizer`; Stage 3
 //! (`plan_selection`, the cost model and its inputs) and the facade (`pass`)
 //! remain here for now.
 //!
@@ -70,8 +71,5 @@ pub use recurrence::{
     UpdateRate,
 };
 
-/// #509 Stage 2 MVP: physical operator implementation, all at query time.
-pub mod physical_candidates;
-
 /// #509 Stage 3 MVP: accuracy check, analytical pricing, cheapest valid plan.
 pub mod plan_selection;

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

@@ -42,7 +42,6 @@ use crate::analytical_cost::{
     estimate_operator, AnalyticalCostError, PhysicalOperator, ResourceCalibration, ResourceEstimate,
 };
 use crate::cost_model::{CostModel, DefaultCostModel};
-use crate::physical_candidates::{stage2_physical, PhysicalCandidate};
 use crate::physical_operator_statistics::{
     EdgeStatistics, OperatorStatistics, PartitionStatistics, UnaryEdgeStatistics,
 };
@@ -53,6 +52,9 @@ use asap_logical_optimizer::pass1::logical_candidates::{
     choice_index, combination_count, compose_logical_candidate, enumerate_choices, nested_targets,
     LocalLogicalCandidates,
 };
+use asap_physical_optimizer::implementation::physical_candidates::{
+    stage2_physical, PhysicalCandidate,
+};
 
 pub const COST_UNIT: &str = "cpu_ms_per_workload_evaluation";
 pub const COST_SOURCE: &str = "analytical-cost-v1 (illustrative statistics)";
@@ -1010,8 +1012,8 @@ fn summary_shape(family: &FieldDataType) -> (u64, u64) {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::physical_candidates::stage2_physical;
     use crate::test_support::lower_promql;
+    use asap_physical_optimizer::implementation::physical_candidates::stage2_physical;
     use asap_types::ir::QueryRoot;
     use asap_types::workload::{Evidence, Rate};
 

crates/integration-tests/Cargo.toml

@@ -9,6 +9,7 @@ asap-frontend-promql = { path = "../frontend-promql" }
 asap-frontend-sql = { path = "../frontend-sql" }
 asap-aware-mapping = { path = "../asap-aware-mapping" }
 asap-logical-optimizer = { path = "../logical-optimizer" }
+asap-physical-optimizer = { path = "../physical-optimizer" }
 
 [dev-dependencies]
 asap-planner = { path = "../planner" }

crates/integration-tests/tests/planner_layering_example1.rs

@@ -61,7 +61,7 @@ mod stages {
         pub label: String,
         pub dag: PhysicalASAPDAG,
         pub query_roots: Vec<LogicalASAPNodeId>,
-        pub stage2: asap_aware_mapping::physical_candidates::PhysicalCandidate,
+        pub stage2: asap_physical_optimizer::implementation::physical_candidates::PhysicalCandidate,
     }
 
     /// Whole-workload cost of one physical candidate; `per_node` has one
@@ -184,8 +184,10 @@ mod stages {
             .enumerate()
             .map(|(index, l)| {
                 let mut stage2 =
-                    asap_aware_mapping::physical_candidates::stage2_physical(&l.id, &l.roots)
-                        .unwrap_or_else(|e| panic!("{}: {e}", l.id));
+                    asap_physical_optimizer::implementation::physical_candidates::stage2_physical(
+                        &l.id, &l.roots,
+                    )
+                    .unwrap_or_else(|e| panic!("{}: {e}", l.id));
                 stage2.id = format!("P{}", index + 1);
                 stage2.label = l.label.clone();
                 PhysicalCandidate {

crates/logical-optimizer/tests/stage1_cost_independence.rs

@@ -7,7 +7,11 @@ use std::path::{Path, PathBuf};
 const FORBIDDEN_MODULES: &[&str] = &["cost_model", "recurrence"];
 
 /// Crates Stage 1 must not depend on: later stages, the facade and the executor.
-const FORBIDDEN_CRATES: &[&str] = &["asap-aware-mapping", "asap-physical-operators"];
+const FORBIDDEN_CRATES: &[&str] = &[
+    "asap-physical-optimizer",
+    "asap-aware-mapping",
+    "asap-physical-operators",
+];
 
 /// Every `.rs` file under `dir`.
 fn rust_files(dir: &Path) -> Vec<PathBuf> {

crates/physical-optimizer/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "asap-physical-optimizer"
+version = "0.1.0"
+edition = "2021"
+
+# #509 Stage 2: physical candidates for each Stage 1 logical candidate.
+# Production code depends only on asap-types: a Stage 1 candidate reaches it as
+# OperatorNode roots. Stage 1 and the PromQL front end are test-only
+# dev-dependencies. Never depends on Stage 3, the facade or the executor.
+# tests/stage2_dependencies.rs checks this manifest.
+[dependencies]
+asap-types = { path = "../types" }
+thiserror = "2"
+
+[dev-dependencies]
+asap-logical-optimizer = { path = "../logical-optimizer" }
+asap-frontend-promql = { path = "../frontend-promql" }

crates/physical-optimizer/src/implementation/mod.rs

@@ -0,0 +1,3 @@
+//! Physical operator implementation of one logical candidate.
+
+pub mod physical_candidates;

crates/physical-optimizer/src/lib.rs

@@ -0,0 +1,15 @@
+//! `asap-physical-optimizer` — #509 Stage 2: physical candidates.
+//!
+//! It turns each Stage 1 logical candidate into physical candidates: how each
+//! operator is implemented, and (once Stage 2 materialization exists) which
+//! sub-DAGs are materialized and when. Cargo enforces the stage order: this
+//! crate depends only on `asap-types`, never on Stage 3, the facade or the
+//! executor.
+//!
+//! - [`implementation`] — physical operator implementation. Every node runs at
+//!   query time until Stage 2 materialization (#509) exists.
+
+pub mod implementation;
+
+#[cfg(test)]
+mod test_support;

crates/physical-optimizer/src/test_support.rs

@@ -0,0 +1,42 @@
+// Fixture helpers for this crate's tests: the subset of
+// `asap-logical-optimizer`'s `test_support` that Stage 2 tests use.
+
+use std::rc::Rc;
+
+use asap_types::ir::OperatorNode;
+use asap_types::types::AccuracyTarget;
+use asap_types::workload::{
+    AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, PlanningWorkload,
+    Predictability, Query, QueryLanguage, QueryRequirements, QueryWorkload, TimeSelection,
+};
+
+pub(crate) fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Rc<OperatorNode> {
+    let workload = PlanningWorkload {
+        query_workload: QueryWorkload {
+            language: QueryLanguage::PromQL,
+            query_batch: Some(vec![BatchEntry {
+                query: Query(query.into()),
+                requirements: QueryRequirements {
+                    accuracy: AccuracyRequirement::Explicit(accuracy),
+                    ..Default::default()
+                },
+                predictability: Predictability::Unknown,
+                invocations: 1,
+                execute_at: None,
+                time_selection: TimeSelection::default(),
+            }]),
+            repeating_queries: None,
+        },
+        data_workload: Some(DataWorkload {
+            data_ingestion_interval: Evidence {
+                value: Some(DurationMs(1_000)),
+                ..Default::default()
+            },
+            ..Default::default()
+        }),
+    };
+    asap_frontend_promql::lower_promql_workload(&workload, 0)
+        .unwrap()
+        .pop()
+        .unwrap()
+}

crates/physical-optimizer/tests/stage2_dependencies.rs

@@ -0,0 +1,29 @@
+//! Stage 2 (physical candidates) depends on no later stage: Cargo enforces
+//! the one-way #509 stage flow (#572).
+
+use std::path::Path;
+
+/// Crates Stage 2 must not depend on: Stage 3, the facade and the executor.
+const FORBIDDEN_CRATES: &[&str] = &[
+    "asap-plan-selection",
+    "asap-aware-mapping",
+    "asap-physical-operators",
+];
+
+/// The manifest names no later stage, facade or executor crate, so Cargo
+/// rejects any import of them.
+#[test]
+fn stage2_manifest_has_no_path_back_to_later_stages() {
+    let manifest =
+        std::fs::read_to_string(Path::new(env!("CARGO_MANIFEST_DIR")).join("Cargo.toml")).unwrap();
+    let offenders: Vec<&str> = manifest
+        .lines()
+        .filter(|line| !line.trim_start().starts_with('#'))
+        .filter(|line| FORBIDDEN_CRATES.iter().any(|name| line.contains(name)))
+        .collect();
+    assert!(
+        offenders.is_empty(),
+        "asap-physical-optimizer must not depend on a later stage:\n{}",
+        offenders.join("\n")
+    );
+}