PhysicalASAPDAG::validate accepts an unstored ingestion-time output read at query time #747

Problem, by example

The #738 shape, as an exported PhysicalASAPDAG (node ids as compile_physical_asap_dag assigns them):

2 SummaryAgg(Kll k=200)   output_state: query_time/summary_state   NotMaterialized
1   Aggregate(max)        output_state: ingestion_time/raw         NotMaterialized
0     Scan(latency_ms)    output_state: ingestion_time/raw         NotMaterialized

The query-time KLL build (node 2) reads node 1's rows, which only exist during ingestion and are not stored.

Today: PhysicalASAPDAG::validate() returns Ok(()). Checked with a throwaway test: time the plan with node 1 MatAtIngestion(Memory) (valid), export it, then set dag.nodes[1].materialization = NotMaterialized and call dag.validate(). The timing pass (apply_materialization_timings) now rejects the same shape before export, with UnstoredIngestionInputUnderQueryTimeSummary (#746). But a DAG built by hand or deserialized from a document skips that pass, and nothing in the export check catches it.

Expected: validate() rejects any edge whose producer runs at ingestion time and is not MatAtIngestion, and whose consumer runs at query time, naming both ids.

Cause

PhysicalASAPDAG::validate (crates/types/src/ir/physical_export.rs) checks the opposite direction (QueryDependencyInIngestion: a query-time producer feeding an ingestion-time consumer), and checks each node's timing against its own materialization (MaterializationTimingMismatch). It never checks that an ingestion-time output read at query time is stored.

Proposal

Add a variant next to QueryDependencyInIngestion, about 15 lines in all:

#[error("query node {consumer:?} reads ingestion node {producer:?}, whose output is not stored (not MatAtIngestion)")]
UnstoredIngestionOutputReadAtQueryTime { producer: PhysicalASAPNodeId, consumer: PhysicalASAPNodeId },

and in the edge loop:

if producer.output_state.timing == ExecutionTiming::IngestionTime
    && nodes[&edge.consumer].output_state.timing == ExecutionTiming::QueryTime
    && producer.materialization.timing() != ExecutionTiming::IngestionTime
{
    return Err(PhysicalASAPDAGValidationError::UnstoredIngestionOutputReadAtQueryTime { .. });
}

Two existing fixtures have this bad shape and need fixing in the same PR:

• crates/types/tests/physical_export.rs:106 (materialization_inconsistent_with_timing_is_rejected): after node 1 is set to MatAtQuery(Disk), the ingestion-time, unstored scan (node 0) feeds it, and dag.validate().unwrap() expects Ok.
• crates/executor/tests/promql_binary.rs:332 (stored_series_evaluations_support_filters_and_sets): the two SummaryMerge nodes (ids 0, 1) are ingestion_time/summary_state with the default NotMaterialized and feed query-time FinalizeExactAccumulator nodes. They should be MatAtIngestion, which is what the test's name ("stored") says.

Scope and acceptance

• In scope: the new variant and check in PhysicalASAPDAG::validate, plus the two fixture fixes.
• Regression test: the exported DAG above returns Err(UnstoredIngestionOutputReadAtQueryTime { producer: 1, consumer: 2 }), and the same DAG with node 1 MatAtIngestion validates. It fails today.
• Planner-exported DAGs are unaffected: since fix(types): reject a query-time summary build over an unstored ingestion-time input #746, the timing pass rejects this shape before export.

Status and related work

Found while fixing #738 (PR #746), where it is listed as a limitation. Related: #738.