[zzylol]

Why

A query-time summary build over an ingestion-time input that nobody stores passed timing validation, although the plan cannot run. At query time that input no longer exists, and the query-time build cannot see the ingestion stream (#738).

SummaryAgg(Kll)        timing: QueryTime,     NotMaterialized
  └─ Aggregate(max)    timing: IngestionTime (fixed by its candidate), NotMaterialized
       └─ Scan(m)

The same holds when the input is ingestion-time exact-accumulator state, e.g. an exact-sum SummaryAgg pinned to ingestion time. Today that case is caught only later, at export, with an unrelated MaterializationTimingMismatch.

What

• validate_asap now rejects a query-time SummaryAgg or PaneChain whose child produces its output at ingestion time (rows or exact-accumulator state) and is not MatAtIngestion. It reports the new error ExecutionDataStateError::UnstoredIngestionInputUnderQueryTimeSummary, which names the edge and the input node's operator.
• The same DAG with the input node assigned MatAtIngestion still validates: the query-time build reads the stored output.
• The build's existing QueryTimeValueUnderIngestionSummary errors now name PaneChain.child for a PaneChain; before, they always said SummaryAgg.child.
• The edge-rule table in the timing module docs gains the new rule for SummaryAgg.child / PaneChain.child.

How

The SummaryAgg.child / PaneChain.child arm of validate_asap (crates/types/src/ir/properties/timing.rs) gets the stored-output check the NonASAP consumer rule already uses (reads_stored: child.materialization.timing() == IngestionTime). The edge label is computed once and shared by the arm's three errors:

let edge = match op {
    ASAPOp::PaneChain { .. } => "PaneChain.child",
    _ => "SummaryAgg.child",
};
// ...
if timing == ExecutionTiming::QueryTime
    && avail.timing == ExecutionTiming::IngestionTime
    && child.materialization.timing() != ExecutionTiming::IngestionTime
{
    return Err(ExecutionDataStateError::UnstoredIngestionInputUnderQueryTimeSummary {
        edge,
        operator: child.operator.kind_name(),
    });
}

Stage 2 never produces this assignment. No production code fixes a node at IngestionTime (only tests do, via with_timing), and a node without a fixed timing either takes its consumer's timing or follows its own MatAtIngestion assignment. So only hand-built or candidate-fixed plans are affected, and no candidate set changes.

Key code interfaces added / modified

crates/types/src/ir/properties/execution_data_state.rs

/// A plan-construction-time data_state violation found by timing validation. The type stays.
pub enum ExecutionDataStateError {
    /// A MaintainPopulation/EvaluatePopulation pair with an invalid contract.
    InvalidMaintainedPopulation, // unchanged
    /// A query-time value under a summary built at ingestion time. `edge` now says
    /// "PaneChain.child" for a PaneChain (was always "SummaryAgg.child").
    QueryTimeValueUnderIngestionSummary { edge: &'static str, child: ExecutionDataState }, // unchanged
    /// A summary build at query time over an input (rows or exact-accumulator state) produced
    /// at ingestion time and not stored (not MatAtIngestion). Bad input: a hand-built or
    /// candidate-fixed plan. `edge` is "SummaryAgg.child" or "PaneChain.child"; `operator` is
    /// the input node's kind. Message: "a query-time {edge} reads {operator} output produced
    /// at ingestion time that is not stored (not MatAtIngestion)".
    UnstoredIngestionInputUnderQueryTimeSummary { edge: &'static str, operator: &'static str }, // new
    /// Any other edge whose child data_state the parent does not accept.
    IllegalChildDataState { edge: &'static str, child: ExecutionDataState }, // unchanged
    /// A SummaryAgg child carrying non-exact-accumulator summary state.
    UnsupportedStateComposition { family: String }, // unchanged
    /// One shared node assigned two different timings by its consumers.
    ConflictingTiming { first: ExecutionDataState, second: ExecutionDataState }, // unchanged
    /// A root of rows that runs at ingestion time without being stored there.
    UnstoredIngestionRoot { operator: &'static str }, // unchanged
    /// An explicit materialization that needs a different timing from the node's fixed one.
    MaterializationConflictsWithFixedTiming { operator: &'static str, fixed: ExecutionTiming, materialization: Materialization }, // unchanged
    /// A filtered summary build placed at ingestion time.
    FilteredSummaryAtIngestion { operator: &'static str }, // unchanged
    /// An ingestion-time BinaryOp the update path cannot evaluate.
    InvalidIngestionTimeBinary, // unchanged
    /// A checked division without one valid guard on a read-time division.
    InvalidCheckedDivision, // unchanged
    /// An exact operator reading a non-plain column.
    NonPlainOperand { column: String, dtype: String }, // unchanged
    /// An ASAP operator with no execution timing contract yet.
    UnimplementedOperator { operator: &'static str }, // unchanged
    /// MatAtIngestion over a sub-DAG that can only run at query time.
    UpstreamNotAtIngestion { operator: &'static str }, // unchanged
    /// A node reached by export without a timing.
    UntimedNode { operator: &'static str }, // unchanged
    /// The exported physical DAG failed its own validation.
    InvalidPhysicalDAG(PhysicalASAPDAGValidationError), // unchanged
}

Usage, from the regression tests in timing.rs (query_time_summary_over_unstored_ingestion_rows_is_rejected, ..._state_is_rejected). Stage 2 calls the same apply_materialization_timings for each materialization candidate:

// rows: max(value) over a scan of metric "m". Its candidate fixed its timing at
// IngestionTime (as a builder does for an operator that must feed maintenance). Not stored.
let rows = placed_at_ingestion(max(scan()));
// summary: a KLL(k=200) build over those rows. It has no assignment, so it is
// NotMaterialized and runs at query time.
let summary = agg(Rc::clone(&rows), kll());
// Timing pass with the default assignment (nothing stored): now rejected.
assert_eq!(
    apply_materialization_timings(&summary, &MaterializationAssignment::default(), &mut TimingMemo::new()).err(),
    Some(ExecutionDataStateError::UnstoredIngestionInputUnderQueryTimeSummary {
        edge: "SummaryAgg.child",
        operator: "Aggregate",
    }),
);
// assignment: stores the rows node at ingestion time (MatAtIngestion(Memory)).
let mut assignment = MaterializationAssignment::default();
assignment.set(&rows, Materialization::MatAtIngestion(Medium::Memory));
// timed: the query-time build now reads the stored rows, so the plan validates.
let timed = apply_materialization_timings(&summary, &assignment, &mut TimingMemo::new()).unwrap();
assert_eq!(timed.timing, Some(ExecutionTiming::QueryTime));

// state: an exact-sum SummaryAgg pinned at ingestion time and not stored, read by
// the query-time KLL build: rejected the same way, with operator "SummaryAgg".
let state = placed_at_ingestion(agg(scan(), exact_sum()));
let summary = agg(Rc::clone(&state), kll());

Before this PR

The rows DAG above validated: apply_materialization_timings returned Ok. The result was a physical plan whose query-time KLL build reads rows that exist only during ingestion. The exact-state variant also passed the timing pass and failed only at export, with MaterializationTimingMismatch.

After this PR

Both fail in the timing pass. The message is a query-time SummaryAgg.child reads Aggregate output produced at ingestion time that is not stored (not MatAtIngestion), or ... reads SummaryAgg output ... for the exact-state case. Storing the input node (MatAtIngestion) makes either plan valid. Stage 2 candidates, Stage 3 selections and the Example 1–4 plans and costs are unchanged.

Verification

• New regression tests query_time_summary_over_unstored_ingestion_rows_is_rejected and query_time_summary_over_unstored_ingestion_state_is_rejected: each failed before its fix (left: None) and passes after.
• cargo nextest run -p asap-types -p asap-physical-asap-optimization -p asap-plan-selection: 435 passed.
• cargo nextest run -p asap-integration-tests on planner_layering_example1..4, materialization_e2e, tumbling_pane_merge, exact_composition, reuse_rules, frequency_execution, stage2_physical_asap_optimization_sql, kll_pane_execution and precompute_raw_samples: 119 passed. The example tests pin the selected plans and costs, so those are unchanged.
• Tests that hand-place timings: asap-frontend-sql maintained_population + pearson_corr (15 passed), asap-executor weighted_topk_binding + promql_binary (7 passed), asap-logical-asap-optimization (178 passed).
• cargo fmt --all -- --check and cargo clippy -p asap-types --all-targets --all-features --locked -- -D warnings: clean.
• The PaneChain.child label on the older QueryTimeValueUnderIngestionSummary errors has no dedicated test: apply_materialization_timings rejects that shape earlier, with UpstreamNotAtIngestion.

Limitations / follow-ups

• PhysicalASAPDAG::validate (the exported-DAG check) has no matching rule. A hand-built or deserialized exported DAG where an unstored ingestion-time producer feeds a query-time consumer still returns Ok(()). Tracked in PhysicalASAPDAG::validate accepts an unstored ingestion-time output read at query time #747.

Closes #738