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

[zzylol]

Why

#509 (Stage 2) hands each candidate plan to the runtime as a PhysicalASAPDAG: the #511 operator DAG plus, for every node, when it runs (ingestion or query time) and what it produces (rows or summary state). #537 exports only the logical DAG, with no timing, and the runtime still compiles the legacy PostAsapDAG built from SummaryExpr. To run plans built from the unified IR (#539, #540), the IR needs a timing pass and a physical export, and the runtime needs a compiler that reads that export.

What

• Timing (asap_types::ir::timing): apply_lifecycle_timings writes an ExecutionTiming into every node of a plan from a per-summary LifecycleAssignment, and validates every edge against the node kinds.
• Physical export (asap_types::ir::physical_export): compile_physical_asap_dag / compile_physical_asap_workload turn a timed plan, or a batch with one root per query, into a validated PhysicalASAPDAG.
• Runtime compiler (asap_physical_operators::unified_physical_planner, unified_sources, expressions::unified_planner): compiles a PhysicalASAPDAG into the existing physical operators, with precompute frontiers for maintained states. It is a reference executor for tests; executing anything it cannot lower natively is the deployment runtime's job.

The planner still builds and runs legacy plans; it switches to these entry points at the cutover (#542).

How

1. Timing pass. The logical DAG carries no timing. LifecycleAssignment records, per SummaryAgg (by node identity), whether its state is maintained at ingestion time or recomputed at query time (default: maintained). apply_lifecycle_timings then times every node:

   • SummaryEstimate / EvaluatePopulation run at query time, MaintainPopulation at ingestion time;
   • a SummaryAgg takes its assignment, unless something below it can only exist at query time (an evaluation, or a state the assignment recomputes at query time), which moves it to query time too;
   • every other node runs when its consumer runs.

   A node reached by two consumers with different timings is a ConflictingTiming error; split_shared_by_phase copies such a sub-DAG first. TimingMemo shares the pass across the roots of a batch, so a node shared by two roots stays one node.

2. Physical export. compile_physical_asap_workload flattens the timed roots into nodes and edges (shared sub-DAGs once), and records each node's ExecutionDataState (timing + raw rows or summary state), each edge's role (Left / Right / Input / ScalarRef) and its grouping/window compatibility. PhysicalASAPDAG::validate checks the roots, that each node's incoming edges are exactly its operator's inputs with their roles, and that no ingestion-time node reads a query-time value. A DAG that fails validation is an InvalidPhysicalDAG error.

3. Runtime compiler. compile(dag, inputs, roots) walks the DAG from the selected roots down to the bound inputs and lowers each node to physical operators. Native lowering covers the relational and summary operators. A node with no native lowering (most PromQL vector expressions) fails compilation with an explicit error: ASAPPlanner generates and costs candidates, and the deployment runtime (e.g. the ASAPQuery backend) executes them, so there is no fallback evaluator in the planner. Node ids above u32::MAX are rejected, because the compiler numbers its helper operators above the Planner id range. Precompute: a frontier of nodes can be materialized (cut_candidate, enumerate_frontiers), and a boundary over a time-series TimeRange reads raw samples, while a table range needs a stored population. bind_with_data_sources resolves Scan leaves through DataSources.

The runtime crate's chrono pin is relaxed to 0.4.39 because DataFusion 54 (#614) needs ^0.4.44.

Key code interfaces

Timing — crates/types/src/ir/timing.rs:

pub struct LifecycleAssignment { /* SummaryAgg identity -> ExecutionTiming */ }
impl LifecycleAssignment {
    pub fn default_maintained() -> Self;
    pub fn set(&mut self, summary: &Rc<OperatorNode>, timing: ExecutionTiming);
    pub fn summary_timing(&self, summary: &Rc<OperatorNode>) -> ExecutionTiming;
}
pub struct TimingMemo { /* input node -> timed node, shared across roots */ }
pub fn apply_lifecycle_timings(root: &Rc<OperatorNode>, assignment: &LifecycleAssignment, memo: &mut TimingMemo)
    -> Result<Rc<OperatorNode>, ExecutionDataStateError>;
pub fn split_shared_by_phase(root: &Rc<OperatorNode>, assignment: &LifecycleAssignment) -> Rc<OperatorNode>;
pub fn data_state(node: &OperatorNode) -> Option<ExecutionDataState>;
pub fn validate_default(root: &Rc<OperatorNode>) -> Result<(), ExecutionDataStateError>;
pub fn planned_data_state(node: &Rc<OperatorNode>) -> Option<ExecutionDataState>;

Physical export — crates/types/src/ir/physical_export.rs:

pub struct PhysicalASAPDAG { pub nodes: Vec<PhysicalASAPDAGNode>, pub edges: Vec<PhysicalASAPDAGEdge>, pub roots: Vec<PhysicalASAPNodeId> }
pub struct PhysicalASAPDAGNode { pub id, pub payload: Operator<NodeId>, pub output_state: ExecutionDataState, pub output_schema, pub guarantee }
pub struct PhysicalASAPDAGEdge { pub producer, pub consumer, pub role: EdgeRole, pub data_state, /* grouping / window compatibility */ }
pub enum EdgeRole { Left, Right, Input, ScalarRef }
impl PhysicalASAPDAG {
    pub fn validate(&self) -> Result<(), PhysicalASAPDAGValidationError>;
    pub fn with_execution_phases(self, /* .. */) -> Result<Self, PhysicalASAPDAGValidationError>;
}
pub fn compile_physical_asap_dag(root: &Rc<OperatorNode>) -> Result<PhysicalASAPDAG, ExecutionDataStateError>;
pub fn compile_physical_asap_workload(roots: &[Rc<OperatorNode>]) -> Result<PhysicalASAPDAG, ExecutionDataStateError>;
pub fn compile_physical_asap_workload_with_node_ids(roots: &[Rc<OperatorNode>])
    -> Result<PhysicalASAPDAGCompilation, ExecutionDataStateError>; // DAG + node identity map

Runtime — crates/asap-physical-operators/src/unified_physical_planner/:

pub fn compile(dag: &PhysicalASAPDAG, inputs: BTreeMap<NodeId, InputContract>, roots: &[NodeId])
    -> Result<CompiledPhysicalDAG, Error>;
pub fn bind<'a>(dag: &PhysicalASAPDAG, sources: BTreeMap<NodeId, Source<'a>>, roots: &[NodeId])
    -> Result<PhysicalDAG<'a, Batch, SchemaRef>, Error>;
pub fn bind_with_data_sources<'a>(dag: &PhysicalASAPDAG, sources: BTreeMap<NodeId, Source<'a>>,
    roots: &[NodeId], data_sources: &DataSources) -> Result<PhysicalDAG<'a, Batch, SchemaRef>, Error>;
pub fn compile_candidate(..) / cut_candidate(..) / enumerate_frontiers(..) / select_candidate(..);

Fields and errors

Item	Meaning
ExecutionDataState { timing, primitive }	When a node runs (IngestionTime / QueryTime) and whether it produces raw rows or summary state.
EdgeRole	Which input slot an edge feeds. Left / Right order a binary operator, Input is a single input, ScalarRef is a plan read from a scalar position.
PhysicalASAPDAGValidationError::NoRoots	Empty batch.
PhysicalASAPDAGValidationError::EdgeRoleMismatch	A node's incoming edges are not exactly its operator's inputs with their roles (for example, swapped Left / Right).
ExecutionDataStateError::ConflictingTiming	A shared node is needed at two timings.
ExecutionDataStateError::UntimedNode	Export of a node the timing pass did not reach.
ExecutionDataStateError::InvalidPhysicalDAG	The exported DAG failed its own validation.
no native binding	compile fails for a node the reference executor cannot lower; the deployment runtime executes such nodes.
InputContract	Schema and properties of a bound input (raw source or a materialized frontier).

Key code interfaces added / modified

crates/types/src/ir/physical_export.rs (new module planner_types::ir::physical_export)

/// Wire version a `PhysicalASAPDAGDocument` must carry; `validate` rejects any other.
pub const PHYSICAL_ASAP_DAG_WIRE_VERSION: u32 = 8;            // new
/// Index of a node in the exported DAG (the flat-DAG `NodeId`).
pub type PhysicalASAPNodeId = NodeId;                          // new
/// A node's operator, with each child replaced by its node id.
pub type PhysicalASAPOperatorPayload = Operator<NodeId>;       // new

/// The timed plan the Planner hands to the runtime: operator nodes, typed
/// producer-to-consumer edges, and one root per query of the batch.
pub struct PhysicalASAPDAG {                                   // new
    /// Every operator node, children before parents.
    pub nodes: Vec<PhysicalASAPDAGNode>,
    /// One edge per operator input, with its role and the state crossing it.
    pub edges: Vec<PhysicalASAPDAGEdge>,
    /// One root per query of the batch, in workload order; scalar query roots are not nodes yet.
    pub roots: Vec<PhysicalASAPNodeId>,
}
/// One operator of the exported DAG with its execution phase, schema and guarantee.
pub struct PhysicalASAPDAGNode {                               // new
    /// This node's id; edges and roots refer to it.
    pub id: PhysicalASAPNodeId,
    /// The operator, with children as node ids.
    pub payload: PhysicalASAPOperatorPayload,
    /// Phase (ingestion or query time) and primitive (rows or summary state) of the output.
    pub output_state: ExecutionDataState,
    /// Schema of the node's output.
    pub output_schema: Schema,
    /// Accuracy guarantee of the output, if the node has one.
    pub guarantee: Option<ResultGuarantee>,
}
/// One input of a consumer, carrying what flows across it.
pub struct PhysicalASAPDAGEdge {                               // new
    /// Node whose output is read.
    pub producer: PhysicalASAPNodeId,
    /// Node that reads it.
    pub consumer: PhysicalASAPNodeId,
    /// Which input slot of the consumer this edge fills.
    pub role: EdgeRole,
    /// The producer's output schema; `validate` checks they match.
    pub intermediate_schema: Schema,
    /// The producer's output state; `validate` checks they match.
    pub data_state: ExecutionDataState,
    /// How the consumer's grouping relates to the producer's (summary-to-summary edges only).
    pub grouping: GroupingEdgeCompatibility,
    /// Whether this edge needs aligned panes (set when both ends run at ingestion time).
    pub window: WindowEdgeCompatibility,
}
/// The consumer input slot an edge fills: the single `Input`, a binary
/// operator's `Left`/`Right`, or `ScalarRef` when read from inside a scalar expression.
pub enum EdgeRole { Input, Left, Right, ScalarRef }
/// Grouping relation of a `SummaryAgg` feeding a `SummaryAgg`: same reduction,
/// consumer groups by a subset, or incompatible; `NotApplicable` for any other edge.
pub enum GroupingEdgeCompatibility { Identical, ConsumerCoarsensProducer, Incompatible, NotApplicable }
/// The first variant marks an ingestion-to-ingestion edge, which physical lowering
/// must align by pane phase or an exact boundary residual; otherwise `NotApplicable`.
pub enum WindowEdgeCompatibility { RequiresAlignedPanePhaseOrExactWindowEdgeResidual, NotApplicable }

/// Versioned JSON hand-off from Planner to runtime.
#[serde(deny_unknown_fields)]
pub struct PhysicalASAPDAGDocument { pub schema_version: u32, pub dag: PhysicalASAPDAG }   // new
/// Why an exported DAG or document is malformed: a wrong wire version, broken
/// ids/edges/roots, a cycle, or a query-time producer under an ingestion-time node.
/// From the Planner it is an internal bug; from a received document it is bad input.
pub enum PhysicalASAPDAGValidationError { /* IncompletePhaseAssignment, QueryDependencyInIngestion { .. },
                                             UnsupportedVersion(u32), DuplicateNodeId(..), MissingRoot(..), ... */ }

impl PhysicalASAPDAGDocument {
    /// Wraps a DAG with the current wire version.
    pub fn new(dag: PhysicalASAPDAG) -> Self;
    /// Checks the version, then the DAG; called at every process boundary.
    pub fn validate(&self) -> Result<(), PhysicalASAPDAGValidationError>;
}
impl PhysicalASAPDAG {
    /// Reassigns every node's phase (all nodes must be named) without changing operators,
    /// updates edge states, and revalidates. Does not prove deployment support.
    pub fn with_execution_phases(&self, phases: &BTreeMap<PhysicalASAPNodeId, ExecutionTiming>)
        -> Result<Self, PhysicalASAPDAGValidationError>;
    /// Checks ids, roots, edges against operator inputs, schemas, states, phases,
    /// acyclicity, reachability and summary schemas.
    pub fn validate(&self) -> Result<(), PhysicalASAPDAGValidationError>;
}

/// An exported DAG plus the map from its node ids back to the planner's `Rc` nodes.
pub struct PhysicalASAPDAGCompilation { pub dag: PhysicalASAPDAG, pub node_ids: PhysicalASAPNodeIdentityMap }  // new
impl PhysicalASAPNodeIdentityMap {
    /// The exported id of a planner node (by `Rc` identity), if it was exported.
    pub fn node_id(&self, node: &Rc<OperatorNode>) -> Option<PhysicalASAPNodeId>;
    /// The planner node exported under `id`.
    pub fn operator_node(&self, id: PhysicalASAPNodeId) -> Option<&Rc<OperatorNode>>;
}

/// Exports the timed DAG below one root; every node must already carry a timing.
pub fn compile_physical_asap_dag(root: &Rc<OperatorNode>) -> Result<PhysicalASAPDAG, ExecutionDataStateError>;           // new
/// Same, also returning the id-to-node map.
pub fn compile_physical_asap_dag_with_node_ids(root: &Rc<OperatorNode>) -> Result<PhysicalASAPDAGCompilation, ExecutionDataStateError>;
/// Exports a timed batch as one DAG with one root per query; shared nodes stay one node.
pub fn compile_physical_asap_workload(roots: &[Rc<OperatorNode>]) -> Result<PhysicalASAPDAG, ExecutionDataStateError>;   // new
/// Same, also returning the id-to-node map.
pub fn compile_physical_asap_workload_with_node_ids(roots: &[Rc<OperatorNode>]) -> Result<PhysicalASAPDAGCompilation, ExecutionDataStateError>;

crates/types/src/ir/timing.rs, post_asap/execution_data_state.rs, post_asap/expr.rs

// new module ir::timing, re-exported from ir
/// Summary materialization's lifecycle choice per `SummaryAgg` (by identity):
/// maintained at ingestion time or recomputed at query time.
pub struct LifecycleAssignment { /* summary node -> ExecutionTiming */ }
impl LifecycleAssignment {
    /// Every summary maintained at ingestion time (the empty assignment).
    pub fn default_maintained() -> Self;
    /// Records the timing chosen for one summary node.
    pub fn set(&mut self, summary: &Rc<OperatorNode>, timing: ExecutionTiming);
    /// The timing chosen for a summary node.
    pub fn summary_timing(&self, summary: &Rc<OperatorNode>) -> ExecutionTiming;   // IngestionTime if unset
}
/// Input-to-timed node memo shared by all roots of one workload, so a node
/// shared by two roots stays one `Rc`.
pub struct TimingMemo { /* .. */ }
impl TimingMemo {
    /// An empty memo, one per workload.
    pub fn new() -> Self;
    /// The timed copy produced for `input`, if the pass reached it.
    pub fn timed(&self, input: &Rc<OperatorNode>) -> Option<&Rc<OperatorNode>>;
}
/// Writes the assignment's timings into every node below `root`, then validates
/// every edge; returns the timed copy of `root`. Run before export.
pub fn apply_lifecycle_timings(root: &Rc<OperatorNode>, assignment: &LifecycleAssignment, memo: &mut TimingMemo)
    -> Result<Rc<OperatorNode>, ExecutionDataStateError>;
/// Planning-time legality check of a candidate under the default assignment,
/// with `root` consumed at `root_timing`; keeps nothing.
pub fn validate_default(root: &Rc<OperatorNode>, root_timing: ExecutionTiming) -> Result<(), ExecutionDataStateError>;
/// The data state a timed node outputs; `None` if the node is untimed.
pub fn data_state(node: &OperatorNode) -> Option<ExecutionDataState>;
/// The data state `node` would output under the default assignment when its consumer runs at `consumer`.
pub fn planned_data_state(node: &Rc<OperatorNode>, consumer: ExecutionTiming) -> ExecutionDataState;
/// Copies, for one consumer, each sub-DAG the assignment would reach with two
/// timings (e.g. a CSE-shared `Scan`), so the workload can be timed.
pub fn split_shared_by_phase(root: &Rc<OperatorNode>, assignment: &LifecycleAssignment) -> Rc<OperatorNode>;

/// Why a plan's execution timing is not executable (an unsatisfiable placement
/// or a malformed plan), returned at planning time rather than at runtime.
pub enum ExecutionDataStateError {
    /// A shared node is reached with two different timings that splitting did not resolve.
    ConflictingTiming { first: ExecutionDataState, second: ExecutionDataState },          // new
    /// Export met a node with no timing; an internal bug (timing pass not run).
    UntimedNode { operator: &'static str },                                               // new
    /// An edge feeds a query-time value or an unaccepted state into a maintained summary; unexecutable plan.
    EvaluationUnderMaintenance { edge: &'static str, child: ExecutionDataState },         // new
    /// The exported DAG failed its own validation, e.g. an empty batch or a hand-timed plan.
    InvalidPhysicalDAG(#[from] PhysicalASAPDAGValidationError),                           // new
    /* existing variants unchanged */
}

impl post_asap::BinaryOperator {
    /// Converts a unified-IR binary operator to the post-ASAP kernel form; a comparison
    /// becomes `CompareBool` when `return_bool` is set.
    pub fn from_logical(operator: &crate::ir::BinaryOperator, return_bool: bool) -> Self;  // new
}

crates/asap-physical-operators/src/unified_physical_planner/mod.rs, compiled.rs, candidates.rs

pub mod unified_physical_planner;   // new: compiles a PhysicalASAPDAG into executable operators

/// A live operator bound at a storage/ingestion frontier; it must have exactly
/// the declared schema and no upstream dependencies.
pub type Source<'a> = Box<dyn PhysicalOperator<Batch, SchemaRef> + 'a>;
/// Lowers the DAG to physical operators against typed input contracts, without opening readers.
pub fn compile(dag: &PhysicalASAPDAG, inputs: BTreeMap<NodeId, InputContract>, roots: &[NodeId])
    -> Result<CompiledPhysicalDAG, Error>;
/// Compiles against the sources' contracts and instantiates with those sources, for callers that already have them.
pub fn bind<'a>(dag: &PhysicalASAPDAG, sources: BTreeMap<NodeId, Source<'a>>, roots: &[NodeId])
    -> Result<PhysicalDAG<'a, Batch, SchemaRef>, Error>;
/// Like `bind`, also resolving raw `Scan` leaves against registered data sources.
pub fn bind_with_data_sources<'a>(dag: &PhysicalASAPDAG, sources: BTreeMap<NodeId, Source<'a>>, roots: &[NodeId],
    data_sources: &crate::unified_sources::DataSources) -> Result<PhysicalDAG<'a, Batch, SchemaRef>, Error>;

/// A typed execution boundary (schema plus plan properties) without storage identity or a live reader.
pub struct InputContract { pub schema: SchemaRef, pub properties: PlanProperties }
impl InputContract {
    /// A bounded input with the given schema and unknown emission.
    pub fn bounded(schema: SchemaRef) -> Self;
    /// The contract a live source declares.
    pub fn from_source(source: &dyn PhysicalOperator<Batch, SchemaRef>) -> Self;
}
/// Selected physical operators and their input slots, serializable; rebinding never lowers again.
pub struct CompiledPhysicalDAG { /* nodes: BTreeMap<NodeId, Node>, roots: Vec<NodeId> */ }
impl CompiledPhysicalDAG {
    /// Links already-compiled fragments; source and fragment ids share one namespace.
    pub fn compose(sources: BTreeMap<NodeId, InputContract>, fragments: BTreeMap<NodeId, (Vec<NodeId>, Self)>, roots: Vec<NodeId>) -> Result<Self, Error>;
    /// Assembles already-lowered operators and typed external inputs.
    pub fn from_operators(inputs: BTreeMap<NodeId, InputContract>, operators: BTreeMap<NodeId, (Vec<NodeId>, Operator)>, roots: Vec<NodeId>) -> Result<Self, Error>;
    /// The output nodes.
    pub fn roots(&self) -> &[NodeId];
    /// The external inputs the DAG expects to be bound.
    pub fn input_contracts(&self) -> impl Iterator<Item = (NodeId, &InputContract)>;
    /// The contract of a reachable node's output, derived without readers.
    pub fn output_contract(&self, id: NodeId) -> Result<InputContract, Error>;
    /// Validates by instantiating against contract-only sources.
    pub fn validate(&self) -> Result<(), Error>;
    /// Binds exactly the declared inputs to live sources and validates before any source starts.
    pub fn instantiate<'a>(&self, sources: BTreeMap<NodeId, Source<'a>>) -> Result<PhysicalDAG<'a, Batch, SchemaRef>, Error>;
}

/// Precompute/query split at a materialization frontier.
pub struct CompiledPhysicalPlan {
    /// Operators up to the frontier, run ahead of queries; `None` for an empty frontier.
    pub precompute: Option<CompiledPhysicalDAG>,
    /// Operators run at query time, reading the materialized outputs.
    pub query: CompiledPhysicalDAG,
    /// Contract of each frontier output stored between the two.
    pub materialized_outputs: BTreeMap<NodeId, InputContract>,
}
/// Compiles the DAG and cuts it at `frontier`; an empty frontier keeps everything in the query DAG.
pub fn compile_candidate(dag: &PhysicalASAPDAG, inputs: BTreeMap<NodeId, InputContract>, roots: &[NodeId], frontier: &[NodeId])
    -> Result<CompiledPhysicalPlan, Error>;
/// Lowers the DAG once and cuts it at each frontier; per-frontier failures stay visible.
pub fn compile_candidates(dag: &PhysicalASAPDAG, inputs: BTreeMap<NodeId, InputContract>, roots: &[NodeId], frontiers: &[Vec<NodeId>])
    -> Vec<Result<CompiledPhysicalPlan, Error>>;
/// Partitions an existing compile result at `frontier` without lowering again.
pub fn cut_candidate(compiled: &CompiledPhysicalDAG, frontier: &[NodeId]) -> Result<CompiledPhysicalPlan, Error>;
/// The frontier the DAG's phases imply: ingestion-time nodes read by query-time
/// nodes, plus an ingestion-timed root. Errors if a query-time node feeds an ingestion-time one.
pub fn frontier_from_timing(dag: &PhysicalASAPDAG) -> Result<Vec<NodeId>, Error>;
/// Enumerates reachable antichain frontiers above the inputs, up to
/// `max_candidates`; exceeding the budget is an error, never a partial list.
pub fn enumerate_frontiers(dag: &PhysicalASAPDAG, inputs: &BTreeMap<NodeId, InputContract>, roots: &[NodeId], max_candidates: usize)
    -> Result<Vec<Vec<NodeId>>, Error>;

/// A candidate's total workload cost over one scope and horizon, supplied by the evaluator.
pub struct CandidateCost { pub workload_scope: String, pub horizon_seconds: f64, pub total_cost: f64 }
/// The winning candidate, its index in the input list, and its cost.
pub struct CandidateSelection<T = CompiledPhysicalPlan> { pub candidate: T, pub candidate_index: usize, pub cost: CandidateCost }
/// Picks the cheapest compiled candidate the evaluator prices (`None` = infeasible);
/// errors if costs differ in scope/horizon, are invalid, or none is feasible.
pub fn select_candidate<T>(candidates: Vec<Result<T, Error>>, evaluate: impl FnMut(&T) -> Result<Option<CandidateCost>, Error>)
    -> Result<CandidateSelection<T>, Error>;

unified_physical_planner/precompute.rs, promql_rows.rs, promql_values.rs

// precompute (new)
/// Compiles a selected precompute sub-DAG whose inputs are already-computed state boundaries.
pub fn compile(dag: &PhysicalASAPDAG, frontiers: &[NodeId], roots: &[NodeId]) -> Result<CompiledPhysicalDAG, Error>;
/// Row schema of a stored population: `$population` labels and `$window_end` alongside the family's state.
pub fn population_schema(family: SummaryFamilyType) -> SchemaRef;
/// The stored-population schema for a logical schema with one summary state; used in installed-plan recovery.
pub fn source_schema(logical: &Schema) -> Result<SchemaRef, Error>;
/* raw_sample_schema, raw_sample_row, boundary_schema, is_population_schema */

// promql_rows (new): PromQL series identity
/// Encodes a full label set as a canonical, reversible JSON string.
pub fn encode_series_identity(labels: &BTreeMap<String, String>) -> Result<String, Error>;
/// Decodes a series identity back to its label set.
pub fn decode_series_identity(encoded: &str) -> Result<BTreeMap<String, String>, Error>;
/// Adds the series-identity column to the plan's PromQL rows; run before candidate search.
pub fn with_series_identity(root: &Rc<OperatorNode>) -> Result<Rc<OperatorNode>, Error>;
/* series_row, compile_current_series_evaluation, compile_rate_ranking, compile_fixed_window_rate_aggregation */

// promql_values (new): value schemas and per-operator builders
/// Physical schema of a PromQL scalar value.
pub fn scalar_schema() -> SchemaRef;
/// Physical schema of a PromQL instant vector, carrying full label sets.
pub fn vector_schema() -> SchemaRef;
/// Physical schema of a PromQL range vector (matrix).
pub fn matrix_schema() -> SchemaRef;
/* compile_scalar, compile_temporal, compile_histogram_quantile, compile_binary, compile_aggregate,
   compile_sort, compile_limit, compile_negate, compile_vector_to_scalar, ... */

crates/asap-physical-operators/src/unified_sources/mod.rs, memory.rs

pub mod unified_sources;   // new: raw data sources for bound plans

/// A connector returning raw, unfiltered rows; each scan opens an independent cursor.
pub trait RawSource {
    /// The rows' schema, stable for the binding's lifetime.
    fn schema(&self) -> SchemaRef;
    /// Declares whether scans are a finite snapshot.
    fn boundedness(&self) -> crate::plan::Boundedness;   // default Unknown
    /// Opens a cursor; a connector error is an error, never an empty scan.
    fn scan(&self, context: RunContext) -> Result<OutputStream<'_, Batch>, Error>;
}
/// Explicitly registered raw sources keyed by Planner `Source` identity; no discovery or fallback.
pub struct DataSources { /* sources: Vec<(Source, Arc<dyn RawSource>)> */ }
impl DataSources {
    /// Registers a source; errors on a duplicate identity or an invalid schema.
    pub fn register(&mut self, identity: Source, source: Arc<dyn RawSource>) -> Result<(), Error>;
    /// Binds a Planner `Scan` leaf to its registered source with compiled predicates;
    /// errors if unbound, the schema differs, or a predicate is not boolean.
    pub fn bind(&self, expression: &OperatorNode) -> Result<Scan, Error>;
}
/// The physical scan operator `bind` returns.
pub struct Scan { /* reader, output, predicates */ }
/// Immutable in-memory raw data, for tests and local runs.
pub struct MemorySource { /* .. */ }
impl MemorySource {
    /// Errors if the schema has summary-state columns or a batch's schema differs.
    pub fn new(schema: SchemaRef, batches: Vec<Batch>) -> Result<Self, Error>;
}

crates/asap-physical-operators/src/expressions/unified_planner.rs, expressions/mod.rs, expressions/binary.rs, operators/joins/mod.rs

pub struct CompiledExpression { /* .. */ }          // new: evaluates unified-IR scalar expressions
impl CompiledExpression {
    /// Type-checks an expression against a plain (non-summary) input schema.
    pub fn compile(expression: &ScalarExpr, input: &SchemaRef) -> Result<Self, Error>;
    /// Evaluates one row of the schema it was compiled against.
    pub fn evaluate(&self, row: &[Value]) -> Result<Value, Error>;
}

/// A physical row expression.
pub enum Expression {
    /// A legacy Planner scalar expression.
    Planner(Box<crate::expressions::CompiledExpression>),
    /// A unified-IR scalar expression.
    UnifiedPlanner(Box<unified_planner::CompiledExpression>),   // new
    /* ... */
}
impl Expression {
    /// Wraps a compiled unified-IR expression.
    pub fn unified_planner(expression: unified_planner::CompiledExpression) -> Self;   // new
}

// expressions::binary (new): temporary kernel aliases
/// Re-exported so kernels keep their paths during the compiler migration.
pub use planner_types::post_asap::BinaryOperator;
/// Re-exported for the same reason.
pub use planner_types::pre_asap::BinaryOpKind;

impl operators::Operator {
    /// A relational join whose predicate is a unified-IR expression over the concatenated left and right schemas.
    pub fn unified_relational_join(left: SchemaRef, right: SchemaRef, kind: JoinKind,
        predicate: &planner_types::ir::Predicate, output: SchemaRef) -> Result<Self, Error>;   // new
}

Usage:

// Planner side: export a timed plan as the wire document.
let dag = compile_physical_asap_workload(&timed_roots)?;
let doc = PhysicalASAPDAGDocument::new(dag);
doc.validate()?;

// Runtime side: split at the timing frontier and compile.
let frontier = frontier_from_timing(&doc.dag)?;
let plan = compile_candidate(&doc.dag, inputs, &doc.dag.roots, &frontier)?;

Before this PR

The timing pass and physical export existed only for the legacy SummaryExpr graph (PostAsapDAG), and the runtime compiled only that graph. A plan built from the unified IR (#539, #540) could be exported as a logical DAG (#537) but not timed, exported for execution, or run.

After this PR

A unified-IR plan can be timed, exported as a validated PhysicalASAPDAG with one root per batch query, and the parts the reference executor supports natively can be compiled and executed, including plans with materialized precompute frontiers. The legacy runtime is unchanged; the planner switches at #542.

Evidence

Execution examples from the tests. Screenshots and measurements: not applicable.

Case	Result	Test
Scan → SummaryAgg(sum by key) → Finalize, default assignment	SummaryAgg at ingestion time, Finalize at query time; the export round-trips through JSON	timed_plan_exports_with_timing
two queries reading one summary state	one SummaryAgg node, two roots	batch_shares_the_summary_and_keeps_one_root_per_query
Left / Right edges swapped	EdgeRoleMismatch	swapped_edge_roles_are_rejected
compile_physical_asap_workload(&[])	InvalidPhysicalDAG(NoRoots), not a panic	an_empty_batch_is_an_error
a state recomputed at query time below another summary	the consuming summary moves to query time	query_time_state_moves_its_consuming_summary_to_query_time
node ids shifted above u32::MAX	Err, not a panic or an id collision	node_ids_above_u32_are_rejected
TimeRange over a table Scan as a precompute boundary	not raw samples; needs a stored population	only_a_time_series_range_reads_raw_samples

Verification

• Unit tests: crates/types/src/ir/timing.rs (timing rules and conflicts), crates/types/tests/physical_export.rs (export, validation, batches, edge roles, empty batch).
• Runtime tests: unified_precompute.rs (raw-sample contract of a precompute boundary), unified_node_ids.rs (ids above u32::MAX are rejected). Native lowering needs more tests: Runtime: tests for the native lowering of PhysicalASAPDAG #662.
• Review fixes: every finding fixed after the review has a test that fails without the fix.
• End-to-end tests: not applicable in this PR; the planner is not cut over until refactor(planner): plan the workload on the unified operator IR #542.
• Other checks: cargo fmt --check, workspace all-target Clippy with -D warnings, and cargo test --workspace (2,014 tests, 0 failures) on the PR head.

Architectural decisions

• Timing is written into the node (OperatorNode.timing) by a separate pass over the logical DAG, not stored beside it, so the export and the runtime read one value per node.
• One PhysicalASAPDAG per batch, one root per query, with shared sub-DAGs exported once, matching the logical export (feat(ir): add logical sub-DAG sharing and export without execution timing #537).
• A parallel unified_physical_planner, promoted to physical_planner when the legacy runtime is removed (refactor(planner): plan the workload on the unified operator IR #542 and the PR after it). Rejected: converting PhysicalASAPDAG into the legacy PostAsapDAG.
• No fallback evaluator. An earlier version evaluated retained exact PromQL sub-DAGs in the planner's runtime. It was removed: the planner's output is a plan choice; executing what the native operators cannot run belongs to the deployment runtime.

Limitations and follow-up

• Validation of per-operator timing in a deserialized PhysicalASAPDAG: PhysicalASAPDAG::validate does not check per-operator timing #656.
• Batches whose queries need different timings for a shared node fail with ConflictingTiming: Timing: a node shared across batch queries with different timings fails with ConflictingTiming #654.
• split_shared_by_phase returns a rebuilt DAG keyed by new node identities, so a non-default assignment has to be re-applied by the caller; Stage 2 materialization: ingestion vs query time per summary #604 (Stage 2 materialization) returns the remapped assignment. Nothing calls it with a non-default assignment before then.
• SQL differences found by the feat(sql): lower SQL to unified operator and scalar IR #539 DataFusion check: sum / avg over empty or all-NULL input return 0 / NaN instead of NULL, Int64 avg can overflow, and several SQL shapes are not executable yet: Runtime: sum/avg over empty input, avg overflow, and unsupported SQL shapes #659.
• The unified runtime duplicates parts of the legacy one (compiled.rs, the value helpers) until refactor(planner): plan the workload on the unified operator IR #542 and the PR after it remove the legacy runtime.

Stack: tracker #528 · previous: #540 · next: #542.

Human review — do not complete with an agent

• The MVP boundary is correct.
• New conceptual layers or public interfaces are necessary.
• The before/after description matches the intended product behavior.
• Human reviewer:
• Decision and rationale:

[zzylol]

Third review and the fallback decision

Decision: ASAPPlanner has no execution fallback. It generates candidates, costs them, or says why a query has none; executing anything the native operators cannot run is the deployment runtime's job (e.g. the ASAPQuery backend). ccc1f83 (refactor(runtime): remove the PromQL fallback evaluator) removes the evaluator (900 lines) and its 45 tests; a node with no native lowering now fails compilation with an explicit error.

#	Finding (third review)	Resolution
1	Rows feeding maintained state at ingestion time were routed through the PromQL fallback.	Gone with the fallback.
2	Node ids above u32::MAX were accepted and could collide with helper ids (debug builds panicked).	Fixed in ccc1f83: Error::Invalid; also an error instead of a debug_assert when one node needs 2^16 helper ids. Test: node_ids_above_u32_are_rejected.
3	promql_fallback::raw_series panicked on ASAP input.	Gone with the fallback.
4	forces_query_time forces query time for an unaligned PromQL BinaryOp, undocumented.	Documented in ccc1f83.
5	Previously serialized join plans may not deserialize.	Not addressed: compatibility with old plans is not required.
6	The native lowering has almost no tests of its own.	#662
7	round(m, scalar(n)) and similar fail in series-identity resolution.	#661

The earlier fixes that lived in the fallback (unary minus __name__, the clamp scalar filter, the binding check below a fallback node) went with it. The clamp Filter { NOT (max < min) } from #540 stays in the IR.