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).

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:

🤖 Generated with Claude Code

https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW

[zzylol]

Review findings and fixes

Two review rounds. All fixes are in c42fd93 (fix(runtime): address #541 review findings), each with a test that fails without the fix.

#	Finding	Resolution
1	Two batch queries sharing a node with different timings fail with ConflictingTiming. Not reachable today: the only workload-timing caller up the stack uses all-query-time.	Not changed: #654.
2	A SummaryAgg recomputed at query time did not move the SummaryAgg consuming it to query time (EvaluationUnderMaintenance).	Fixed: forces_query_time reads the assignment.
3	PhysicalASAPDAG::validate did not check edge roles, so swapped Left / Right would compute b - a.	Fixed: incoming edges must be exactly the operator's inputs with their roles (EdgeRoleMismatch).
4	Precompute treated any TimeRange as raw PromQL samples.	Fixed: only a TimeRange over a time-series scan; a table range needs a stored population.
5	An input binding nothing reads was silently ignored.	Fixed for the harmful case: a binding below a PromQL fallback node (which reads raw series slots instead) is an error. A binding below another bound input is still superseded by it, as in cut; rejecting it broke per-frontier recompilation up the stack.
6	raw_series_contract_is_explicit asserted only is_ok() and its comment said the opposite of the code.	Fixed.
7	promql_rows shape guard dropped (low confidence).	Not changed.
8	(follows #540) The fallback kept __name__ for unary minus.	Fixed: dropped, as Prometheus 3 does.
9	(round 2) split_shared_by_phase rebuilds nodes under new identities, so a non-default assignment no longer applies to them.	Not changed here: nothing calls it with a non-default assignment until #604, which returns the remapped assignment.
10	(round 2) The export panicked (expect) on a DAG that failed validation, e.g. an empty batch.	Fixed: ExecutionDataStateError::InvalidPhysicalDAG.
11	(round 2) validate does not check each operator kind's timing. Only compiler-produced DAGs exist today.	Not changed: #656.
12	(round 2) Dead scalar-literal branch in the unified planner.	Fixed: removed.
13	(round 2) The unified runtime duplicates parts of the legacy one.	Intended: the legacy runtime is removed in #542 and the unified one moves to physical_planner in the PR after it.
14	(#540 round 2) clamp with max < min must return no series.	The fallback now runs #540's Filter { NOT (max < min) } (a filter on scalars alone) instead of filtering inside the projection.

The DataFusion check of #539 also ran lowered SQL through this runtime: sum / avg over empty or all-NULL input, avg overflow on Int64, and unsupported SQL shapes are #659.

[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.