feat(runtime): export timed plans as PhysicalASAPDAG and compile them

crates/asap-physical-operators/Cargo.toml

@@ -4,6 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 futures = "0.3"
 planner-types = { package = "asap-types", path = "../types" }
 asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "5f03ccbd798ed5fec62bdd839bcb331123cab369" }

crates/asap-physical-operators/README.md

@@ -54,6 +54,9 @@ 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
+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.

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

@@ -0,0 +1,3 @@
+//! Temporary kernel aliases during the unified compiler migration.
+pub use planner_types::post_asap::BinaryOperator;
+pub use planner_types::pre_asap::BinaryOpKind;

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

@@ -5,7 +5,9 @@ use crate::{
 };
 use planner_types::pre_asap::{ArithmeticOpKind, DataType};
 pub mod arithmetic;
+pub mod binary;
 mod planner;
+pub mod unified_planner;
 pub use planner::CompiledExpression;
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug)]
 pub enum Expression {
@@ -15,6 +17,7 @@ pub enum Expression {
         right: Box<Expression>,
     },
     Planner(Box<crate::expressions::CompiledExpression>),
+    UnifiedPlanner(Box<unified_planner::CompiledExpression>),
     Column(usize),
     ExactFloat64(usize),
     FiniteFloat64(Box<Expression>),
@@ -53,6 +56,9 @@ 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: crate::expressions::CompiledExpression) -> Self {
         Self::Planner(Box::new(expression))
     }
@@ -99,6 +105,10 @@ impl Expression {
                 };
                 Ok((dtype, n || m))
             }
+            UnifiedPlanner(expression) => {
+                expression.validate_input(input)?;
+                Ok(expression.dtype())
+            }
             Planner(expression) => {
                 expression.validate_input(input)?;
                 Ok(expression.dtype())
@@ -286,6 +296,7 @@ impl Expression {
                 }
             }
             Planner(expression) => expression.evaluate(row)?,
+            UnifiedPlanner(expression) => expression.evaluate(row)?,
             Label { column, name } => {
                 let Value::Map(entries) = &row[*column] else {
                     return Err(invalid("label read requires a map"));

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

@@ -336,10 +336,6 @@ pub struct CompiledExpression {
     output: (DataType, bool),
 }
 impl CompiledExpression {
-    pub(crate) fn expression(&self) -> &QueryExpr {
-        &self.expression
-    }
-
     pub fn compile(expression: &QueryExpr, input: &SchemaRef) -> Result<Self, Error> {
         let schema = input
             .fields