From eec9a3fa329d2607220a4745c69f102189da2948 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 5 Nov 2026 15:06:54 +0000 Subject: [PATCH] feat(types): an unmaterialized summary build runs at its consumer's timing, including ingestion A summary build or MaintainPopulation that is not materialized and has no assignment entry now runs where its consumer runs: in the query run, or inside the ingestion run of the materialized state that reads it ((IngestionTime, None)). A filtered build, or one over something that only exists at query time, still runs at query time. Export validation checks every edge with readable_at: a query-time node reading an unmaterialized ingestion-time output is the new UnmaterializedIngestionOutputReadAtQueryTime; the old UnmaterializedSummaryBuildNotAtQueryTime is deleted. Stage 2 does not offer the state yet, so plans are unchanged. Export reports every edge where satisfies_physical fails as one variant, PhysicalASAPDAGValidationError::PhysicalRequirementNotSatisfied, replacing QueryDependencyInIngestion and the unmaterialized-ingestion- output case. The tests find the build and merge ids inline. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01K1M6br9PL4SDeVS1XpezKJ --- crates/executor/tests/every_valid_state.rs | 86 ++++++++--- .../tests/tumbling_window_merge.rs | 91 +++++++++++ .../src/catalog/configurations.rs | 8 +- crates/plan-selection/src/lib.rs | 3 +- crates/types/src/ir/physical_export.rs | 64 ++++---- crates/types/src/ir/properties/timing.rs | 146 ++++++++++++------ crates/types/tests/physical_export.rs | 52 ++++--- crates/types/tests/schema_rebuilding.rs | 47 ++++++ crates/types/tests/summary_merge_structure.rs | 10 +- docs/design_docs/concepts/post-asap-ir.md | 22 ++- 10 files changed, 385 insertions(+), 144 deletions(-) diff --git a/crates/executor/tests/every_valid_state.rs b/crates/executor/tests/every_valid_state.rs index 42c7c8817..3bc942209 100644 --- a/crates/executor/tests/every_valid_state.rs +++ b/crates/executor/tests/every_valid_state.rs @@ -18,7 +18,7 @@ use planner_types::ir::{ }; use planner_types::physical::{Lifetime, PhysicalProperties}; -/// `FinalizeExactAccumulator(SummaryAgg(Sum)(TimeRange 5 min(Scan latency_ms)))`. +/// `FinalizeExactAccumulator(SummaryMerge(SummaryAgg(Sum)(TimeRange 5 min(Scan latency_ms))))`. fn plan() -> Rc { let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { source: Source::TimeSeries { @@ -50,19 +50,30 @@ fn plan() -> Rc { filter: None, })) .unwrap(); + let merge = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { + children: vec![state], + newest_tumbling_windows: None, + group_by: None, + })) + .unwrap(); OperatorNode::new_shared(Operator::ASAP(ASAPOp::FinalizeExactAccumulator { - child: state, + child: merge, })) .unwrap() } +/// The build under the merge under `root`. +fn build_of(root: &Rc) -> &Rc { + root.children()[0].children()[0] +} + /// What the physical planner makes of the build in each valid state: a /// state materialized across evaluations (either timing, either medium) is -/// a materialized view slot the deployment binds; every other state is -/// computed in the query run (a `OneEvaluation` result once for the run's -/// readers). A summary build not materialized at ingestion time is not -/// built by rule: the timing pass runs an unmaterialized build at query -/// time. +/// a materialized view slot the deployment binds; a query-time state that +/// is not materialized across evaluations is computed in the query run (a +/// `OneEvaluation` result once for the run's consumers). Not materialized at +/// ingestion time, the build sits under the merge materialized at ingestion +/// time and is computed in that view's ingestion run. #[test] fn physical_planner_over_every_valid_state() { use ExecutionTiming::{IngestionTime, QueryTime}; @@ -83,17 +94,38 @@ fn physical_planner_over_every_valid_state() { { let root = plan(); let mut assignment = PhysicalPropertiesAssignment::default(); - let unmaterialized_at_ingestion = - (properties.timing(), properties.lifetime()) == (IngestionTime, Lifetime::None); - if !unmaterialized_at_ingestion { - assignment.set(root.children()[0], properties); + match (properties.timing(), properties.lifetime()) { + // The build gets no entry: it runs in the ingestion run of the + // merge above it, which is materialized at ingestion time. + (IngestionTime, Lifetime::None) => assignment.set( + root.children()[0], + PhysicalProperties::new( + ExecutionTiming::IngestionTime, + Lifetime::AcrossEvaluations(Medium::Memory), + ) + .expect("a valid timing and lifetime"), + ), + _ => assignment.set(build_of(&root), properties), } let timed = apply_physical_properties(&root, &assignment).unwrap(); let mut dag = compile_physical_asap_dag(&root, &timed).unwrap(); let views = dag.set_retention(&HashMap::new(), &HashMap::new()).unwrap(); - let compiled = asap_executor::physical_planner::compile(&dag, &views, &[3]) - .unwrap_or_else(|error| panic!("{properties:?}: {error}")); - let build = compiled.operator_name(2); + let id_of = |summary_merge: bool| { + dag.nodes + .iter() + .find(|node| match &node.payload { + Operator::ASAP(ASAPOp::SummaryAgg { .. }) => !summary_merge, + Operator::ASAP(ASAPOp::SummaryMerge { .. }) => summary_merge, + _ => false, + }) + .expect("the plan has a summary build and a merge") + .id + }; + let (build_id, merge_id) = (id_of(false), id_of(true)); + let compiled = + asap_executor::physical_planner::compile(&dag, &views, &[dag.roots[0] as u64]) + .unwrap_or_else(|error| panic!("{properties:?}: {error}")); + let build = compiled.operator_name(build_id as u64); let computed_in_the_query_run = || { assert_ne!(build, Some("MaterializedView"), "{properties:?}"); assert!(build.is_some(), "{properties:?}"); @@ -108,16 +140,24 @@ fn physical_planner_over_every_valid_state() { (QueryTime, Lifetime::None) => computed_in_the_query_run(), (QueryTime, Lifetime::OneEvaluation(_)) => computed_in_the_query_run(), (IngestionTime, Lifetime::None) => { - // Left unassigned: an unmaterialized build lands at query time. + let node = dag.nodes.iter().find(|node| node.id == build_id).unwrap(); + assert_eq!( + PhysicalProperties::new(node.output_state.timing, node.lifetime), + Ok(properties) + ); + // The query reads the merge's view; its ingestion run + // computes the build. assert_eq!( - PhysicalProperties::new( - dag.nodes[2].output_state.timing, - dag.nodes[2].lifetime - ), - Ok( - PhysicalProperties::new(ExecutionTiming::QueryTime, Lifetime::None) - .expect("a valid timing and lifetime") - ) + compiled.operator_name(merge_id as u64), + Some("MaterializedView") + ); + let refresh = + asap_executor::physical_planner::compile(&dag, &[], &[merge_id as u64]) + .unwrap(); + let build = refresh.operator_name(build_id as u64); + assert!( + build.is_some_and(|name| name != "MaterializedView"), + "{build:?}" ); } (IngestionTime, Lifetime::OneEvaluation(_)) => { diff --git a/crates/integration-tests/tests/tumbling_window_merge.rs b/crates/integration-tests/tests/tumbling_window_merge.rs index d17270b45..ef842725b 100644 --- a/crates/integration-tests/tests/tumbling_window_merge.rs +++ b/crates/integration-tests/tests/tumbling_window_merge.rs @@ -911,3 +911,94 @@ fn queries_at_60_and_90_seconds_read_one_chain_of_30_second_tumbling_windows() { ); } } + +/// End to end, a summary build not materialized at ingestion time: a merge +/// materialized at ingestion time reads the 5-minute Sum build, which is not +/// materialized and so runs inside the merge's ingestion run. The query +/// reads the merge's view and answers as the whole build at query time. +#[test] +fn an_unmaterialized_build_runs_in_the_ingestion_run_of_the_state_reading_it() { + let samples = samples(1); + let single = single_build("sum_over_time(m[5m])", AccuracyTarget::exact()); + let build = summary_agg(&single); + let merged = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { + children: vec![Rc::clone(&build)], + newest_tumbling_windows: None, + group_by: None, + })) + .unwrap(); + let root = replace(&single, &build, &merged); + let mut assignment = PhysicalPropertiesAssignment::default(); + assignment.set( + &merged, + PhysicalProperties::new( + ExecutionTiming::IngestionTime, + Lifetime::AcrossEvaluations(Medium::Memory), + ) + .expect("a valid timing and lifetime"), + ); + let mut exported = export(&root, &assignment); + let id_of = |summary_merge: bool| { + exported + .dag + .nodes + .iter() + .find(|node| match &node.payload { + Operator::ASAP(ASAPOp::SummaryAgg { .. }) => !summary_merge, + Operator::ASAP(ASAPOp::SummaryMerge { .. }) => summary_merge, + _ => false, + }) + .expect("the plan has a summary build and a merge") + .id + }; + let (build_id, merge_id) = (id_of(false), id_of(true)); + let (build_id, merge_id) = (build_id as u64, merge_id as u64); + let properties = |id: u64| { + let node = exported + .dag + .nodes + .iter() + .find(|node| node.id as u64 == id) + .unwrap(); + PhysicalProperties::new(node.output_state.timing, node.lifetime) + }; + assert_eq!( + properties(build_id), + Ok( + PhysicalProperties::new(ExecutionTiming::IngestionTime, Lifetime::None) + .expect("a valid timing and lifetime") + ) + ); + assert_eq!( + properties(merge_id), + Ok(PhysicalProperties::new( + ExecutionTiming::IngestionTime, + Lifetime::AcrossEvaluations(Medium::Memory) + ) + .expect("a valid timing and lifetime")) + ); + exported.dag.validate().unwrap(); + let views = views_at(&mut exported.dag, &[merge_id]).unwrap(); + // The merge's ingestion run builds the Sum from the raw samples and + // materializes only the merge. + let ingestion = compile(&exported.dag, &[], &[merge_id]).unwrap(); + let rows = run( + &ingestion, + raw_store(&exported.dag, &samples), + Scope::Ingestion { + window_start_ms: 0, + window_end_ms: EVALUATION_MS, + }, + ); + let store = InMemoryDataStore::default() + .with_materialized_view( + merge_id as usize, + vec![Batch::try_new(schema_of(&exported.dag, merge_id), rows).unwrap()], + ) + .unwrap(); + let query = compile(&exported.dag, &views, &[exported.dag.roots[0] as u64]).unwrap(); + assert_eq!( + render(&run(&query, store, query_scope())), + render(&execute(&single, &samples)) + ); +} diff --git a/crates/physical-asap-optimization/src/catalog/configurations.rs b/crates/physical-asap-optimization/src/catalog/configurations.rs index ea980e080..68cb2c1e1 100644 --- a/crates/physical-asap-optimization/src/catalog/configurations.rs +++ b/crates/physical-asap-optimization/src/catalog/configurations.rs @@ -151,8 +151,12 @@ pub fn enumerate_candidate_configurations( // `Disk` state given to it executes in memory (a view slot bound to // the deployment's in-memory store, a one-evaluation result materialized // within the query run), while cost prices it as disk; - // - (IngestionTime, None) for a summary build: an unmaterialized build - // runs at query time (`apply_physical_properties`); + // - (IngestionTime, None) for a summary build: every consumer + // handles it (the build runs inside the ingestion run of the + // materialized state reading it), but offering it adds candidates + // and can change selected plans, which is a separate decision; + // `down_closed` requires every build below an ingestion-time build + // to be materialized there; // - (IngestionTime, AcrossEvaluations) for a non-summary node: rows // are not maintained at ingestion time, only summary states; // - (QueryTime, AcrossEvaluations) for anything but a tumbling-window diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 63262babe..eaa3df1ae 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -5063,8 +5063,7 @@ mod tests { /// The KLL DAG whose node has `properties`: the build itself, except /// for (IngestionTime, None), which is the scan below the build - /// materialized at ingestion time (a summary build that is not - /// materialized runs at query time). + /// materialized at ingestion time. fn kll_dag_with(properties: PhysicalProperties) -> PhysicalASAPDAG { match (properties.timing(), properties.lifetime()) { (ExecutionTiming::IngestionTime, Lifetime::None) => kll_dag(at_ingestion()), diff --git a/crates/types/src/ir/physical_export.rs b/crates/types/src/ir/physical_export.rs index 61ae190a0..0a61fee10 100644 --- a/crates/types/src/ir/physical_export.rs +++ b/crates/types/src/ir/physical_export.rs @@ -26,11 +26,12 @@ use crate::ir::properties::execution_data_state::{ use crate::ir::properties::guarantee::AccuracyGuarantee; #[cfg(any(test, feature = "test-support"))] use crate::ir::properties::timing::apply_physical_properties_to_roots; -use crate::ir::properties::timing::{ - data_state, is_materializable_state_output, PhysicalPropertiesAssignment, -}; +use crate::ir::properties::timing::{data_state, PhysicalPropertiesAssignment}; use crate::ir::schema::{FieldDataType, Schema}; -use crate::physical::{InvalidPhysicalProperties, Lifetime, PhysicalProperties}; +use crate::physical::{ + satisfies_physical, Evaluation, InvalidPhysicalProperties, Lifetime, PhysicalProperties, + PhysicalRequirement, +}; /// A node's operator, with each child replaced by its node id. pub type PhysicalASAPOperatorPayload = Operator; @@ -179,14 +180,12 @@ pub enum PhysicalASAPDAGValidationError { #[source] reason: InvalidPhysicalProperties, }, - /// A summary build (or maintained population) that is not materialized but does - /// not run at query time: a planner bug. - #[error( - "node {node:?} is a summary build that is not materialized but does not run at query time" - )] - UnmaterializedSummaryBuildNotAtQueryTime { node: PhysicalASAPNodeId }, - #[error("ingestion node {consumer:?} depends on query node {producer:?}")] - QueryDependencyInIngestion { + /// An edge where α^phys fails (`satisfies_physical` is false): ingestion + /// work reads a query-time output, or a query-time node reads an + /// ingestion-time output that is not materialized (it runs only inside + /// the ingestion run that reads it). A planner bug. + #[error("node {consumer:?} cannot read node {producer:?} at its timing")] + PhysicalRequirementNotSatisfied { producer: PhysicalASAPNodeId, consumer: PhysicalASAPNodeId, }, @@ -604,13 +603,13 @@ impl PhysicalASAPDAG { pub fn validate(&self) -> Result<(), PhysicalASAPDAGValidationError> { let mut nodes = HashMap::new(); + let mut properties_of = HashMap::new(); for node in &self.nodes { if nodes.insert(node.id, node).is_some() { return Err(PhysicalASAPDAGValidationError::DuplicateNodeId(node.id)); } - // A node's timing and lifetime must go together; a summary - // build that is not materialized runs at query time. Any other node that - // is not materialized runs when its consumer runs. + // A node's timing and lifetime must go together. A node that is + // not materialized runs when its consumer runs (the edge checks below). let properties = PhysicalProperties::new(node.output_state.timing, node.lifetime) .map_err( |reason| PhysicalASAPDAGValidationError::InvalidPhysicalProperties { @@ -618,17 +617,7 @@ impl PhysicalASAPDAG { reason, }, )?; - if is_materializable_state_output(&node.payload) - && properties - == PhysicalProperties::new(ExecutionTiming::IngestionTime, Lifetime::None) - .expect("a valid timing and lifetime") - { - return Err( - PhysicalASAPDAGValidationError::UnmaterializedSummaryBuildNotAtQueryTime { - node: node.id, - }, - ); - } + properties_of.insert(node.id, properties); let materialized_across_evaluations = match (properties.timing(), properties.lifetime()) { ( @@ -750,13 +739,22 @@ impl PhysicalASAPDAG { edge.consumer, )); } - if producer.output_state.timing == ExecutionTiming::QueryTime - && nodes[&edge.consumer].output_state.timing == ExecutionTiming::IngestionTime - { - return Err(PhysicalASAPDAGValidationError::QueryDependencyInIngestion { - producer: edge.producer, - consumer: edge.consumer, - }); + // α^phys: a node reads, in the same evaluation, only a producer + // that satisfies its timing (`satisfies_physical`). + let requirement = PhysicalRequirement::new( + nodes[&edge.consumer].output_state.timing, + Evaluation::Same, + None, + ) + .expect("a same-evaluation read is valid at either timing"); + let properties = properties_of[&edge.producer]; + if !satisfies_physical(&properties, &requirement) { + return Err( + PhysicalASAPDAGValidationError::PhysicalRequirementNotSatisfied { + producer: edge.producer, + consumer: edge.consumer, + }, + ); } if edge.intermediate_schema != producer.output_schema { return Err(PhysicalASAPDAGValidationError::EdgeSchemaMismatch { diff --git a/crates/types/src/ir/properties/timing.rs b/crates/types/src/ir/properties/timing.rs index 3529faf93..26df18918 100644 --- a/crates/types/src/ir/properties/timing.rs +++ b/crates/types/src/ir/properties/timing.rs @@ -14,11 +14,13 @@ //! //! - `SummaryEstimate` and `EvaluatePopulation` run at query time; //! - a node the assignment lists runs at its entry's timing with its entry's -//! lifetime. A `SummaryAgg` or `MaintainPopulation` the assignment does -//! not list is not materialized and runs at query time. Every node -//! upstream of an ingestion-time node also runs at ingestion time, so an -//! ingestion-time entry over something that can only exist at query time -//! is an error ([`ExecutionDataStateError::UpstreamNotAtIngestion`]); +//! lifetime. Every node upstream of an ingestion-time node also runs at +//! ingestion time, so an ingestion-time entry over something that can only +//! exist at query time (an evaluation, a filtered summary build) is an +//! error ([`ExecutionDataStateError::UpstreamNotAtIngestion`]); +//! - a filtered summary build the assignment does not list runs at query +//! time (raw samples carry none of the columns its filter reads), and so +//! does one over something that only exists at query time; //! - a PromQL `BinaryOp` over per-series inputs that are not the same series //! rows runs at query time, since maintenance cannot align them; //! - every other node is not materialized (lifetime `None`) and runs when its @@ -72,9 +74,11 @@ use crate::physical::{ /// output is stored"). Stage 2 lists each materialized sub-DAG output, named /// by the node whose output is materialized; an entry whose lifetime is /// `None` fixes only the node's timing. A node the assignment does not list -/// is placed by [`apply_physical_properties`]' rules: not materialized, a -/// summary build or `MaintainPopulation` at query time, every other node -/// when its consumer runs. [`apply_physical_properties`] returns the +/// is placed by [`apply_physical_properties`]' rules: not materialized, at +/// query time if it is a filtered summary build or over something that only +/// exists at query time, otherwise when its consumer runs (a summary build +/// under a state materialized at ingestion time runs in that state's +/// ingestion run). [`apply_physical_properties`] returns the /// completed assignment, which lists every node it reached; physical stages /// read a node's timing and lifetime from it. `Default` lists nothing. /// @@ -262,8 +266,11 @@ fn is_unaligned_series_binary(node: &OperatorNode) -> bool { } /// Whether the sub-DAG below `node` contains a node that can only run at -/// query time (an evaluation, or a sub-DAG `assignment` does not put at -/// ingestion time), which forces every consumer above it to query time as well. +/// query time (an evaluation, a filtered summary build, or a node +/// `assignment` places at query time), which forces every consumer above +/// it to query time as well. A node that `assignment` does not list takes +/// its consumer's timing, so it forces query time only when something below +/// it does. /// A PromQL `BinaryOp` whose inputs are not the same per-series rows also runs /// at query time. fn forces_query_time( @@ -287,9 +294,12 @@ fn forces_query_time( Some(listed) if listed.timing() == ExecutionTiming::QueryTime => true, // An ingestion-time entry the DAG cannot honor runs at query time. Some(_) => is_filtered_build(node) || upstream(seen), - // A summary build that is not materialized runs at query time. - None if is_materializable_state_output(&node.operator) => true, + // A filtered build reads columns raw samples do not carry, so it + // runs at query time. + None if is_filtered_build(node) => true, None if is_unaligned_series_binary(node) => true, + // Not listed: it runs where its consumer runs, unless something + // below it cannot run at ingestion time. None => upstream(seen), } }; @@ -424,15 +434,23 @@ fn placement( } return Ok(listed); } - // A summary build that is not materialized runs at query time. - Ok( - if runs_at_query_time_only(node) || is_materializable_state_output(&node.operator) { - PhysicalProperties::new(ExecutionTiming::QueryTime, Lifetime::None) - .expect("a valid timing and lifetime") - } else { - PhysicalProperties::new(consumer, Lifetime::None).expect("a valid timing and lifetime") - }, - ) + // A filtered summary build reads columns raw samples do not carry, and a + // summary build over something that only exists at query time (an + // evaluation) runs there too. Otherwise a node with no entry runs where + // its consumer runs: in the query run, or in the ingestion run of the + // materialized state that reads it. + let query_time_build = is_materializable_state_output(&node.operator) + && (is_filtered_build(node) + || node + .children() + .iter() + .any(|child| forces_query_time(child, assignment, forced))); + Ok(if runs_at_query_time_only(node) || query_time_build { + PhysicalProperties::new(ExecutionTiming::QueryTime, Lifetime::None) + .expect("a valid timing and lifetime") + } else { + PhysicalProperties::new(consumer, Lifetime::None).expect("a valid timing and lifetime") + }) } /// The timing `node` takes when its consumer runs at `consumer`, by the @@ -904,6 +922,16 @@ mod tests { } fn agg(child: Rc, family: FieldDataType) -> Rc { + agg_with(child, family, None) + } + + /// A `SummaryAgg` of `family` over `child`'s sample values, with + /// `filter` (a filtered build never runs at ingestion time). + fn agg_with( + child: Rc, + family: FieldDataType, + filter: Option, + ) -> Rc { std::rc::Rc::new( OperatorNode::with_schema( crate::ir::Operator::ASAP(ASAPOp::SummaryAgg { @@ -912,7 +940,7 @@ mod tests { input: SummaryUpdate::column(ColumnRef::SampleValue), reduction: Reduction::by(vec![]), grouping: GroupingStrategy::default(), - filter: None, + filter, }), Schema::lifted(vec![Field::new("state", family, false)], None), ) @@ -1059,11 +1087,11 @@ mod tests { } } - /// #509 Algorithm 3, line 13: an explicit assignment materialized at ingestion time over a state - /// that only exists at query time is an error, not a silent move to - /// query time. + /// A state materialized at ingestion time over a summary build that is + /// not materialized: the build runs where its consumer runs, inside the + /// ingestion run that refreshes the materialized state. #[test] - fn ingestion_time_node_over_query_time_upstream_is_rejected() { + fn an_unmaterialized_build_under_an_ingestion_time_state_runs_at_ingestion() { let inner = agg(scan(), exact_sum()); let outer = agg(Rc::clone(&inner), kll()); let mut assignment = PhysicalPropertiesAssignment::default(); @@ -1075,21 +1103,36 @@ mod tests { ) .expect("a valid timing and lifetime"), ); + let timed = apply_physical_properties(&estimate(Rc::clone(&outer)), &assignment).unwrap(); assert_eq!( - apply_physical_properties(&estimate(outer), &assignment).err(), - Some(ExecutionDataStateError::UpstreamNotAtIngestion { - operator: "SummaryAgg" - }) + timed.get(&outer), + Some( + PhysicalProperties::new( + ExecutionTiming::IngestionTime, + Lifetime::AcrossEvaluations(Medium::Memory) + ) + .expect("a valid timing and lifetime") + ) ); - assignment.set( - &inner, - PhysicalProperties::new( - ExecutionTiming::IngestionTime, - Lifetime::AcrossEvaluations(Medium::Memory), + assert_eq!( + timed.get(&inner), + Some( + PhysicalProperties::new(ExecutionTiming::IngestionTime, Lifetime::None) + .expect("a valid timing and lifetime") ) - .expect("a valid timing and lifetime"), ); + } + + /// A filtered build cannot run at ingestion time, materialized or not: a + /// state materialized at ingestion time over it names the missing upstream. + #[test] + fn ingestion_time_state_over_a_filtered_build_is_rejected() { + let always = crate::ir::Predicate(crate::ir::ScalarExpr::Literal( + crate::ir::scalar::expr_ir::ScalarValue::Boolean(true), + )); + let inner = agg_with(scan(), exact_sum(), Some(always)); let outer = agg(Rc::clone(&inner), kll()); + let mut assignment = PhysicalPropertiesAssignment::default(); assignment.set( &outer, PhysicalProperties::new( @@ -1098,15 +1141,11 @@ mod tests { ) .expect("a valid timing and lifetime"), ); - let physical_properties = - apply_physical_properties(&estimate(Rc::clone(&outer)), &assignment).unwrap(); - assert_eq!( - lifetime_of(&physical_properties, &outer), - MATERIALIZED_ACROSS - ); assert_eq!( - lifetime_of(&physical_properties, &inner), - MATERIALIZED_ACROSS + apply_physical_properties(&estimate(outer), &assignment).err(), + Some(ExecutionDataStateError::UpstreamNotAtIngestion { + operator: "SummaryAgg" + }) ); } @@ -1221,11 +1260,13 @@ mod tests { ); } - /// A finalize fixed at ingestion time over a state that is not - /// materialized at ingestion time names the missing upstream, not an - /// illegal edge. + /// A finalize fixed at ingestion time (as fixed-window rate candidates + /// build it) over a state that is not materialized: the state runs in the + /// ingestion run with it, and a query-time build cannot read the + /// finalize, which is not stored. #[test] - fn ingestion_fixed_finalize_over_query_time_state_is_upstream_not_at_ingestion() { + fn ingestion_pinned_finalize_over_an_unmaterialized_state_does_not_satisfy_a_query_time_consumer( + ) { let finalize = OperatorNode::new_shared(crate::ir::Operator::ASAP( ASAPOp::FinalizeExactAccumulator { child: agg(scan(), exact_sum()), @@ -1238,9 +1279,12 @@ mod tests { &fixed_at_ingestion(&[&finalize]), ) .err(), - Some(ExecutionDataStateError::UpstreamNotAtIngestion { - operator: "FinalizeExactAccumulator" - }) + Some( + ExecutionDataStateError::UnstoredIngestionInputUnderQueryTimeSummary { + edge: "SummaryAgg.child", + operator: "FinalizeExactAccumulator" + } + ) ); } diff --git a/crates/types/tests/physical_export.rs b/crates/types/tests/physical_export.rs index c04ffa78f..5582197a6 100644 --- a/crates/types/tests/physical_export.rs +++ b/crates/types/tests/physical_export.rs @@ -100,13 +100,13 @@ fn query_time_input_to_ingestion_is_rejected() { *dag.edges[0].data_state_mut() = dag.nodes[0].output_state; assert!(matches!( dag.validate(), - Err(PhysicalASAPDAGValidationError::QueryDependencyInIngestion { .. }) + Err(PhysicalASAPDAGValidationError::PhysicalRequirementNotSatisfied { .. }) )); } /// A node's timing and lifetime must go together: an ingestion-time output -/// cannot be materialized for one evaluation, and a summary build that is not materialized -/// runs at query time. +/// cannot be materialized for one evaluation. A query reads an ingestion-time output +/// only where it is materialized. #[test] fn invalid_physical_properties_are_rejected() { let mut dag = exported_at_ingestion(); @@ -121,7 +121,12 @@ fn invalid_physical_properties_are_rejected() { dag.nodes[1].lifetime = Lifetime::None; assert_eq!( dag.validate(), - Err(PhysicalASAPDAGValidationError::UnmaterializedSummaryBuildNotAtQueryTime { node: 1 }) + Err( + PhysicalASAPDAGValidationError::PhysicalRequirementNotSatisfied { + producer: 1, + consumer: 2 + } + ) ); dag.nodes[1].lifetime = ON_DISK; dag.validate().unwrap(); @@ -771,11 +776,9 @@ fn workload_export_matches_the_single_root_export() { } /// Export validation over every valid state of `plan()`'s summary build: -/// each materialized state, any medium, is assigned, exported and valid; not -/// materialized at query time is assigned and valid; not materialized at -/// ingestion time is valid for the scan below an ingestion-time build and, -/// for the build itself, rejected by rule (an ingestion-time state must be -/// materialized for a query to read it). +/// each state, any medium, is exported and valid. Not materialized at +/// ingestion time is the build under its consumer materialized at ingestion +/// time, computed in that consumer's ingestion run. #[test] fn export_validation_over_every_valid_state() { use ExecutionTiming::{IngestionTime, QueryTime}; @@ -797,13 +800,23 @@ fn export_validation_over_every_valid_state() { let root = plan(); let build = root.children()[0]; let mut assignment = PhysicalPropertiesAssignment::default(); - let unmaterialized_at_ingestion = - (properties.timing(), properties.lifetime()) == (IngestionTime, Lifetime::None); - if !unmaterialized_at_ingestion { - assignment.set(build, properties); + match (properties.timing(), properties.lifetime()) { + // The build gets no entry: it runs in the ingestion run of the + // finalize above it, which is materialized at ingestion time. + (IngestionTime, Lifetime::None) => assignment.set( + &root, + PhysicalProperties::new( + ExecutionTiming::IngestionTime, + Lifetime::AcrossEvaluations(Medium::Memory), + ) + .expect("a valid timing and lifetime"), + ), + // The build gets no entry and runs at query time with its consumer. + (QueryTime, Lifetime::None) => {} + _ => assignment.set(build, properties), } let timed = apply_physical_properties(&root, &assignment).unwrap(); - let mut dag = compile_physical_asap_dag(&root, &timed).unwrap(); + let dag = compile_physical_asap_dag(&root, &timed).unwrap(); let [scan, build, _finalize] = [0, 1, 2].map(|i| { PhysicalProperties::new(dag.nodes[i].output_state.timing, dag.nodes[i].lifetime) }); @@ -813,15 +826,8 @@ fn export_validation_over_every_valid_state() { dag.validate().unwrap(); } (IngestionTime, Lifetime::None) => { - dag.nodes[1].output_state.timing = IngestionTime; - assert_eq!( - dag.validate(), - Err( - PhysicalASAPDAGValidationError::UnmaterializedSummaryBuildNotAtQueryTime { - node: 1 - } - ) - ); + assert_eq!(build, Ok(properties)); + dag.validate().unwrap(); } (IngestionTime, Lifetime::AcrossEvaluations(_)) => { assert_eq!(build, Ok(properties)); diff --git a/crates/types/tests/schema_rebuilding.rs b/crates/types/tests/schema_rebuilding.rs index 8d23fc81a..a70a6bfc7 100644 --- a/crates/types/tests/schema_rebuilding.rs +++ b/crates/types/tests/schema_rebuilding.rs @@ -5,6 +5,8 @@ use asap_types::ir::scalar::ColumnRef; use asap_types::ir::schema::{DataType, Field, FieldDataType, Schema}; use asap_types::ir::schema::{ExactKind, ExactParams, GroupingStrategy, SummaryUpdate}; use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode}; +use asap_types::physical::Lifetime; +use asap_types::physical::Medium; use std::rc::Rc; fn scan(key_type: DataType, name: &str) -> Rc { @@ -402,3 +404,48 @@ fn malformed_population_dags_fail_closed() { spec.metric = "b".into(); assert!(compile_physical_asap_dag(&Rc::new(bad)).is_err()); } + +/// A population that is not materialized, read by a node materialized at +/// ingestion time, runs where its consumer runs: inside that consumer's +/// ingestion run, as (IngestionTime, None). +#[test] +fn an_unmaterialized_population_under_an_ingestion_time_consumer_runs_at_ingestion() { + use asap_types::ir::{apply_physical_properties, PhysicalPropertiesAssignment}; + use asap_types::physical::PhysicalProperties; + let candidate = current_series_topk(); + let Some(ASAPOp::EvaluatePopulation { + child: population, .. + }) = candidate.asap() + else { + unreachable!() + }; + // A key filter keeps whole population states. + let consumer = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Filter { + pred: asap_types::ir::Predicate(asap_types::ir::ScalarExpr::Literal( + asap_types::ir::scalar::expr_ir::ScalarValue::Boolean(true), + )), + child: Rc::clone(population), + })) + .unwrap(); + assert_eq!( + consumer.result_kind, + asap_types::ir::OperatorResultKind::State + ); + let mut assignment = PhysicalPropertiesAssignment::default(); + assignment.set( + &consumer, + PhysicalProperties::new( + ExecutionTiming::IngestionTime, + Lifetime::AcrossEvaluations(Medium::Memory), + ) + .expect("a valid timing and lifetime"), + ); + let timed = apply_physical_properties(&consumer, &assignment).unwrap(); + assert_eq!( + timed.get(population), + Some( + PhysicalProperties::new(ExecutionTiming::IngestionTime, Lifetime::None) + .expect("a valid timing and lifetime") + ) + ); +} diff --git a/crates/types/tests/summary_merge_structure.rs b/crates/types/tests/summary_merge_structure.rs index 877e19503..6434ca50b 100644 --- a/crates/types/tests/summary_merge_structure.rs +++ b/crates/types/tests/summary_merge_structure.rs @@ -733,11 +733,15 @@ fn summary_merge_timing() { timings(&stored_tumbling_windows), (QueryTime, vec![IngestionTime, IngestionTime]) ); - // A `SummaryAgg` without an assignment runs at query time, so a merge - // stored at ingestion time needs its tumbling windows there too. + // A `SummaryAgg` without an assignment is not materialized and runs + // where its consumer runs: under a merge stored at ingestion time, its + // tumbling windows are built in the merge's ingestion run. let mut only_merge = PhysicalPropertiesAssignment::default(); only_merge.set(&root, at_ingestion); - assert!(apply_physical_properties(&root, &only_merge).is_err()); + assert_eq!( + timings(&only_merge), + (IngestionTime, vec![IngestionTime, IngestionTime]) + ); let mut stored_merge = PhysicalPropertiesAssignment::materialized_at_ingestion_where_allowed( std::slice::from_ref(&root), ); diff --git a/docs/design_docs/concepts/post-asap-ir.md b/docs/design_docs/concepts/post-asap-ir.md index 9377eee5d..0ae3f3922 100644 --- a/docs/design_docs/concepts/post-asap-ir.md +++ b/docs/design_docs/concepts/post-asap-ir.md @@ -130,13 +130,18 @@ stages read them from the assignment. lifetime for every node, top-down, and returns the completed assignment: - `SummaryEstimate` and `EvaluatePopulation` run at query time; -- a node the assignment lists runs at its entry's timing; a `SummaryAgg` - or `MaintainPopulation` that is not materialized runs at query time. Every node +- a node the assignment lists runs at its entry's timing. Every node upstream of an ingestion-time node also runs at ingestion time, so a materialized ingestion-time output over something that can only exist at query time (an - evaluation, or a state not materialized at ingestion time) fails with + evaluation, or a filtered summary build) fails with `ExecutionDataStateError::UpstreamNotAtIngestion` ("upstream of ingestion-time node not at ingestion time"); +- a summary build or `MaintainPopulation` the assignment does not list is not + materialized and runs where its consumer runs, like any other node: in the + query run, or inside the ingestion run of the materialized state that reads + it ((IngestionTime, None)). A filtered build always runs at query time, since + raw samples carry none of the columns its filter reads, and so does a build + over something that only exists at query time (an evaluation); - every other node is not materialized and runs when its consumer runs: everything that feeds an ingestion-time node runs at ingestion time, everything above an evaluation at query time. A query-time consumer reads the stored output @@ -200,10 +205,13 @@ children as edges and no embedded sub-DAGs: Each node also records its `lifetime`, including the medium, so the deployment knows which outputs to materialize, for how long and where; validation rejects a timing and lifetime that cannot go together (an ingestion-time -output materialized for `OneEvaluation`) and a `SummaryAgg` or `MaintainPopulation` -that is not materialized outside query time -(`PhysicalASAPDAGValidationError::InvalidPhysicalProperties`). Ingestion -work cannot depend on a future query result. The +output materialized for `OneEvaluation`, +`PhysicalASAPDAGValidationError::InvalidPhysicalProperties`). Every edge is +checked with `satisfies_physical` (α^phys): ingestion work cannot depend +on a future query result, and a query-time node cannot read an +ingestion-time output that is not materialized, which exists only inside +the ingestion run that reads it; either fails with +`PhysicalRequirementNotSatisfied`. The deployment executes the plan as given and does not change its lifetime or timing ([#509 Assumption 8](../proposals/planner-layering.md#assumptions)); it must separately check that it has an implementation and a valid data source for