diff --git a/crates/integration-tests/tests/materialization_e2e.rs b/crates/integration-tests/tests/materialization_e2e.rs index bcce2c805..b342f3a16 100644 --- a/crates/integration-tests/tests/materialization_e2e.rs +++ b/crates/integration-tests/tests/materialization_e2e.rs @@ -117,8 +117,8 @@ fn floats(result: &[Vec]) -> Vec { } /// Execute a timed DAG over one raw-source batch per `Scan` -/// (`batch(schema, metric)`): each materialized view refreshed in an -/// ingestion run over `ingestion`, then the query at `evaluation_time_ms` +/// (`batch(schema, metric)`): each refresh group of materialized views +/// refreshed in one ingestion run over `ingestion`, then the query at `evaluation_time_ms` /// reading the views' states. Returns the root's values and the views. fn execute( dag: &PhysicalASAPDAG, @@ -138,12 +138,16 @@ fn execute( batch(schema, metric).to_rows() }) }; - // Each view's ingestion run refreshes it from the raw data. + // Each refresh group's ingestion run refreshes its views from the raw + // data, compiled once with all of them as roots. let mut store = raw(&dag); - for view in &views { - let refresh = compile(&dag, &[], &[view.node_id as u64]).unwrap(); - let states = physical_common::execute(&refresh, raw(&dag), ingestion.clone()).remove(0); - store = store.with_materialized_view(view.node_id, states).unwrap(); + for group in dag.ingestion_refresh_groups().unwrap() { + let roots: Vec = group.views.iter().map(|view| view.node_id as u64).collect(); + let refresh = compile(&dag, &[], &roots).unwrap(); + let states = physical_common::execute(&refresh, raw(&dag), ingestion.clone()); + for (view, states) in group.views.iter().zip(states) { + store = store.with_materialized_view(view.node_id, states).unwrap(); + } } let query = compile(&dag, &views, &[dag.roots[0] as u64]).unwrap(); let result = physical_common::execute(&query, store, Scope::Query { evaluation_time_ms }); diff --git a/crates/planner/src/installable.rs b/crates/planner/src/installable.rs index 9a63e56ec..20ccc1c65 100644 --- a/crates/planner/src/installable.rs +++ b/crates/planner/src/installable.rs @@ -29,7 +29,9 @@ use std::rc::Rc; use asap_cost::{NodePriceSource, OperatorKind}; use asap_logical_asap_optimization::DeploymentCapabilities; use asap_plan_selection::{SearchReport, Selection}; -use asap_types::ir::materialized_view::{InstallableDagError, MaterializedView, RawDataRetention}; +use asap_types::ir::materialized_view::{ + IngestionRefreshGroup, InstallableDagError, MaterializedView, RawDataRetention, +}; use asap_types::ir::operator::pane_chain::expand_pane_chains_in_workload; use asap_types::ir::physical_export::{ PhysicalASAPDAG, PhysicalASAPDAGEdge, PhysicalASAPDAGNode, PhysicalASAPDAGValidationError, @@ -191,7 +193,8 @@ pub enum InstallablePlanError { #[error("pane retention for node {node_id}, which is not a pane chain of the selected plan")] PaneRetentionOfUnknownChain { node_id: PhysicalASAPNodeId }, /// Unsatisfiable request: the executor cannot refresh materialized view - /// `node_id` from its inputs (`reason`). + /// `node_id` from its inputs (`reason`); for a refresh group, `node_id` + /// is its first view and `reason` names every view compiled with it. #[error("the executor cannot refresh materialized view {node_id} from its inputs: {reason}")] MaterializedViewNotExecutable { node_id: PhysicalASAPNodeId, @@ -300,6 +303,15 @@ impl InstallablePlan { Ok(self.dag.materialized_views()?) } + /// The views refreshed at ingestion, by refresh group, each group after + /// the groups it reads ([`PhysicalASAPDAG::ingestion_refresh_groups`]): + /// a runtime refreshes each group in one ingestion run per window. + pub fn ingestion_refresh_groups( + &self, + ) -> Result, InstallablePlanError> { + Ok(self.dag.ingestion_refresh_groups()?) + } + /// How far back each raw-data source is kept for query-time scans /// ([`PhysicalASAPDAG::raw_data_retention`]): the raw data a runtime /// keeps, the raw-data counterpart of each view's retention. @@ -326,8 +338,10 @@ impl InstallablePlan { } /// Whether the Planner's executor compiles this plan as a runtime runs - /// it: each view's refresh over the views it reads and the `Scan`s it - /// reads (from the runtime's data store), then the query run + /// it: each refresh group's views compiled together (every root its + /// ingestion runs may have) and each other view's refresh, + /// over the views it reads and the `Scan`s it reads (from the runtime's + /// data store), then the query run /// over the views it reads and the `Scan`s it reads outside them. The /// DAG's own `TimeRange`s are compiled, so a window the executor cannot /// run fails here. `Err` holds the executor's error, or why the views @@ -446,12 +460,12 @@ impl<'a> IndexedDag<'a> { } } - /// The sub-DAG of `root` and every node below it, with their edges, in - /// the DAG's node and edge order, rooted at `root`: all a refresh of - /// `root` reads, at the cost of its size rather than the plan's. - fn dag_below(&self, root: PhysicalASAPNodeId) -> PhysicalASAPDAG { + /// The sub-DAG of `roots` and every node below them, with their edges, + /// in the DAG's node and edge order, rooted at `roots`: all a refresh of + /// `roots` reads, at the cost of its size rather than the plan's. + fn dag_below(&self, roots: &[PhysicalASAPNodeId]) -> PhysicalASAPDAG { let mut below = HashSet::new(); - let mut pending = vec![root]; + let mut pending = roots.to_vec(); while let Some(id) = pending.pop() { if below.insert(id) { pending.extend(self.inputs.get(&id).into_iter().flatten().copied()); @@ -467,7 +481,7 @@ impl<'a> IndexedDag<'a> { PhysicalASAPDAG { nodes: ids.iter().map(|id| self.nodes[id].clone()).collect(), edges, - roots: vec![root], + roots: roots.to_vec(), scalar_roots: Vec::new(), } } @@ -503,10 +517,16 @@ fn reader_span_ms_by_node( spans } -/// Whether the executor compiles `dag` as a runtime runs it: each of -/// `views`' refreshes ([`executor_refreshes`]), the first in the whole plan, -/// which runs the executor's checks of the whole plan once, every other over -/// the nodes below it; then the query run ([`executor_runs_queries`]). +/// Whether the executor compiles `dag` as a runtime runs it: each refresh +/// ([`executor_refreshes`]): one per refresh group of the views refreshed +/// at ingestion ([`PhysicalASAPDAG::ingestion_refresh_groups_of`] `views`) +/// and one per other view of `views` (refreshed on first read), the first +/// in the whole plan, which runs the executor's checks of the whole plan +/// once, every other over the nodes below it; then the query run +/// ([`executor_runs_queries`]). A group is compiled with all its views as +/// roots: a compile check covering every run's roots (the views whose +/// state for the run's window is not stored yet, a subset), not the root +/// set of any one run. fn executor_compiles( dag: &PhysicalASAPDAG, views: &[MaterializedView], @@ -514,8 +534,22 @@ fn executor_compiles( let indexed = IndexedDag::new(dag); let views_by_node: HashMap = views.iter().map(|view| (view.node_id, view)).collect(); - for (position, view) in views.iter().enumerate() { - executor_refreshes(view, &views_by_node, &indexed, position == 0)?; + let groups = dag.ingestion_refresh_groups_of(views); + let refreshed_at_ingestion: HashSet = groups + .iter() + .flat_map(|group| group.views.iter().map(|view| view.node_id)) + .collect(); + let refreshes = groups + .iter() + .map(|group| group.views.iter().collect::>()) + .chain( + views + .iter() + .filter(|view| !refreshed_at_ingestion.contains(&view.node_id)) + .map(|view| vec![view]), + ); + for (position, refreshed) in refreshes.enumerate() { + executor_refreshes(&refreshed, &views_by_node, &indexed, position == 0)?; } executor_runs_queries(&indexed, &views_by_node) } @@ -550,41 +584,49 @@ fn executor_runs_queries( }) } -/// Whether the executor compiles `view`'s refresh with the views it reads -/// (among `views_by_node`) as materialized view slots, as a runtime runs it; -/// the executor's reason when it does not. +/// Whether the executor compiles the refresh of `refreshed` (one view, or +/// every view of one refresh group) as roots, with the views they read +/// (among `views_by_node`) that are not among them as materialized view +/// slots; the executor's reason, naming the roots, when it does not. fn executor_refreshes( - view: &MaterializedView, + refreshed: &[&MaterializedView], views_by_node: &HashMap, indexed: &IndexedDag<'_>, in_whole_plan: bool, ) -> Result<(), InstallablePlanError> { use asap_executor::physical_planner::compile; + let roots: Vec = refreshed.iter().map(|view| view.node_id).collect(); // The refresh reads only the nodes below it: compile them, or the whole // plan once, for the executor's checks of the whole plan. let sub_dag; let below = match in_whole_plan { true => indexed.dag, false => { - sub_dag = indexed.dag_below(view.node_id); + sub_dag = indexed.dag_below(&roots); &sub_dag } }; #[cfg(test)] tests::COMPILED_NODES.with(|count| count.set(count.get() + below.nodes.len())); // Raw data is read at each `Scan`, a raw data slot `compile` makes. - let read: Vec = view - .inputs - .materialized_views + let root_set: HashSet = roots.iter().copied().collect(); + let mut read_ids = HashSet::new(); + let read: Vec = refreshed .iter() + .flat_map(|view| &view.inputs.materialized_views) + .filter(|id| !root_set.contains(id) && read_ids.insert(**id)) .filter_map(|id| views_by_node.get(id).map(|read| (*read).clone())) .collect(); - compile(below, &read, &[view.node_id as u64]) + let root_ids: Vec = roots.iter().map(|&id| id as u64).collect(); + compile(below, &read, &root_ids) .map(|_| ()) .map_err( |error| InstallablePlanError::MaterializedViewNotExecutable { - node_id: view.node_id, - reason: error.to_string(), + node_id: roots[0], + reason: match roots.as_slice() { + [_] => error.to_string(), + _ => format!("refreshed with views {roots:?}: {error}"), + }, }, ) } @@ -759,10 +801,11 @@ mod tests { /// A p99 panel over a 300 s window every second, with no raw data kept: /// one DDSketch per series in each of 300 one-second panes built at /// ingestion, about 300 materialized views in a DAG of about 900 nodes. - /// Each view's refresh is compiled over the nodes below it, and the - /// whole plan's checks run once, so the nodes handed to the executor - /// grow with the plan, not with the plan times its views (before, every - /// refresh compiled the whole plan). The count + /// Each refresh (here one refresh group of all the panes) is compiled + /// over the nodes below its views, and the whole plan's checks run once, + /// so the nodes handed to the executor grow with the plan, not with the + /// plan times its views (before, every refresh compiled the whole + /// plan). The count /// covers both `plan_installable`'s build and `check_compiles`. #[tokio::test] async fn view_refreshes_compile_only_the_nodes_below_them() { diff --git a/crates/planner/src/lib.rs b/crates/planner/src/lib.rs index e22160fcb..8ee932ea9 100644 --- a/crates/planner/src/lib.rs +++ b/crates/planner/src/lib.rs @@ -85,8 +85,8 @@ pub use asap_plan_selection::{ AcceptedAccuracy, ErrorSelectingPlan, RejectionCause, SearchReport, SearchStatus, Selection, }; pub use asap_types::ir::materialized_view::{ - InstallableDagError, MaterializedView, MaterializedViewInputs, MaterializedViewRefresh, - RawDataRetention, RawDataWindow, ViewCoverage, + IngestionRefreshGroup, InstallableDagError, MaterializedView, MaterializedViewInputs, + MaterializedViewRefresh, RawDataRetention, RawDataWindow, RefreshWindow, ViewCoverage, }; pub use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPNodeId}; pub use asap_types::ir::properties::summary_coverage::{ColumnIdentity, Constraint, SelectionBox}; diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 0f3af89c1..416a1de65 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -4,21 +4,27 @@ //! the compiler checks that no internal is used (CI runs //! `cargo check -p asap-planner --tests --locked` on its own for that). +use std::cell::Cell; use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use crate::time_series; use asap_executor::physical_planner::compile; -use asap_executor::runtime::{Limits, RunContext, Scope}; +use asap_executor::runtime::{Limits, OutputStream, RunContext, Scope}; +use asap_executor::sources::{ + DataStore, InMemoryDataStore, MaterializedViewRequest, RawDataRequest, +}; use asap_executor::values::{Batch, Value}; +use asap_executor::Error as ExecutorError; use asap_physical_asap_optimization::test_support::split_shared_by_timing; use asap_planner::{ plan_installable, run_planning, CostInputs, DeploymentCapabilities, FrontendInput, InstallablePlan, InstallablePlanError, MaterializedView, MaterializedViewRefresh, OperatorKind, OptimizationInput, OptimizationPass, OptimizeError, PhysicalASAPDAG, PlanOutput, - PlanSearchSettings, PlanningModels, PlanningWorkload, SelectedPlanReport, SelectionBox, - Stage1ToStage3Pass, UserInput, + PlanSearchSettings, PlanningModels, PlanningWorkload, RefreshWindow, SelectedPlanReport, + SelectionBox, Stage1ToStage3Pass, UserInput, }; +use asap_types::ir::operator::Source; use asap_types::ir::schema::schema_support::with_promql_series_identity; use asap_types::ir::schema::{FieldDataType, SketchAlgorithm, SketchParams}; use asap_types::ir::{ @@ -99,6 +105,35 @@ fn remote_write_samples() -> Vec { samples } +thread_local! { + /// How many `read_raw_data` calls the runs on this thread made. + static RAW_DATA_READS: Cell = const { Cell::new(0) }; +} + +/// A store counting its `read_raw_data` calls in [`RAW_DATA_READS`]. +struct CountingStore(InMemoryDataStore); + +impl DataStore for CountingStore { + fn read_raw_data( + &self, + source: &Source, + request: &RawDataRequest, + context: RunContext, + ) -> Result, ExecutorError> { + RAW_DATA_READS.with(|reads| reads.set(reads.get() + 1)); + self.0.read_raw_data(source, request, context) + } + + fn read_materialized_view( + &self, + view: &MaterializedView, + request: &MaterializedViewRequest, + context: RunContext, + ) -> Result, ExecutorError> { + self.0.read_materialized_view(view, request, context) + } +} + /// Compile `ids` of `dag` over the stored states `views` (by node id) and /// run them in `scope` over the raw `samples`, as a runtime does with the /// executor: each id's output batches. @@ -129,7 +164,9 @@ fn run( for (id, batch) in views { store = store.with_materialized_view(id as _, vec![batch]).unwrap(); } - let instance = compiled.instantiate(Arc::new(store)).unwrap(); + let instance = compiled + .instantiate(Arc::new(CountingStore(store))) + .unwrap(); block_on(async { // The run's scope (ingestion window or query time) and limits. let context = RunContext::new(scope, Limits::default()).unwrap(); @@ -256,6 +293,8 @@ struct DataPlaneRun { /// How many materialized views this evaluation refreshed (ingestion or /// query runs); the others it read came from the store. outputs_built: usize, + /// How many `read_raw_data` calls those refreshes made. + raw_data_reads_refreshing: usize, } /// One summary state this test stores: its refreshed state, and how long @@ -313,8 +352,9 @@ fn query_roots(plan: &InstallablePlan) -> Vec { /// The data plane at evaluation time `evaluation_ms`: from `plan_json`, /// `samples` and the states earlier evaluations left in `store`, refresh each -/// materialized view the store lacks once, in the plan's order, from its -/// raw-data windows and the views it reads; answer every query; drop the +/// materialized view the store lacks once, from its raw-data windows and the +/// views it reads: the views refreshed at ingestion by refresh group, one +/// ingestion run per group and span, then the others; answer every query; drop the /// states past their retention; and answer the queries again with the /// query-time reference. fn run_data_plane( @@ -363,9 +403,9 @@ fn run_data_plane_on( let mut inputs = BTreeMap::new(); // How many materialized views this evaluation refreshes. let mut outputs_built = 0; - // Each view's state and the time span of its data at this + // Each view's state key and the time span of its data at this // evaluation, `(start_ms, end_ms]`, by node id. - let mut built: BTreeMap = BTreeMap::new(); + let mut planned: BTreeMap = BTreeMap::new(); for view in &views { // The evaluation its windows end at: this one, or for a tumbling // at-ingestion view the newest point of its grid at or before it. @@ -390,31 +430,81 @@ fn run_data_plane_on( view.inputs .materialized_views .iter() - .map(|id| built[&(*id as u64)].1), + .map(|id| planned[&(*id as u64)].1), ) .reduce(|(a, b), (c, d)| (a.min(c), b.max(d))) .expect("a materialized view reads something"); - let key = state_key(view, built_for_ms); - if !store.contains_key(&key) { - // The views its refresh reads; its raw data is bound at its `Scan`s. - let mut sources = BTreeMap::new(); - for id in &view.inputs.materialized_views { - sources.insert(*id as u64, built[&(*id as u64)].0.clone()); - } - // The run that refreshes it: an ingestion run over its data's span - // once its inputs are complete (each window ends at the run's - // end), or the query run of this evaluation. - let scope = match view.refresh() { - MaterializedViewRefresh::AtIngestion => Scope::Ingestion { - window_start_ms: span.0, - window_end_ms: span.1, - }, - MaterializedViewRefresh::OnFirstRead => Scope::Query { - evaluation_time_ms: evaluation_ms, - }, + planned.insert(view.node_id as u64, (state_key(view, built_for_ms), span)); + } + // The runs that refresh the views the store lacks: for each refresh + // group, one ingestion run per span its views' data covers, with every + // view of the group over that span as a root (a node they share runs + // once); then each view refreshed on first read in this evaluation's + // query run. + let mut refreshes: Vec<(Vec<&MaterializedView>, Scope)> = Vec::new(); + for group in installed.ingestion_refresh_groups().unwrap() { + let mut by_span: BTreeMap<(i64, i64), Vec<&MaterializedView>> = BTreeMap::new(); + for member in &group.views { + let view = views + .iter() + .find(|view| view.node_id == member.node_id) + .unwrap(); + by_span + .entry(planned[&(view.node_id as u64)].1) + .or_default() + .push(view); + } + for (span, members) in by_span { + let scope = Scope::Ingestion { + window_start_ms: span.0, + window_end_ms: span.1, }; - // The batches that run produced, in one batch, as the store keeps it. - let batches = run(dag, sources, samples, &[view.node_id as u64], scope).remove(0); + refreshes.push((members, scope)); + } + } + for view in views + .iter() + .filter(|view| view.refresh() == MaterializedViewRefresh::OnFirstRead) + { + let scope = Scope::Query { + evaluation_time_ms: evaluation_ms, + }; + refreshes.push((vec![view], scope)); + } + let raw_data_reads_before = RAW_DATA_READS.with(Cell::get); + for (members, scope) in refreshes { + // The views of the run whose state the store lacks, one per state. + let mut roots: Vec<&MaterializedView> = Vec::new(); + for view in members { + let key = &planned[&(view.node_id as u64)].0; + if !store.contains_key(key) + && !roots + .iter() + .any(|root| &planned[&(root.node_id as u64)].0 == key) + { + roots.push(view); + } + } + if roots.is_empty() { + continue; + } + // The views the run reads that it does not refresh, from the + // store; its raw data is bound at its `Scan`s. + let mut sources = BTreeMap::new(); + for id in roots + .iter() + .flat_map(|view| &view.inputs.materialized_views) + { + if !roots.iter().any(|root| root.node_id == *id) { + let key = &planned[&(*id as u64)].0; + sources.insert(*id as u64, store[key].state.clone()); + } + } + let root_ids: Vec = roots.iter().map(|view| view.node_id as u64).collect(); + // Each root's batches, in one batch per root, as the store keeps it. + let outputs = run(dag, sources, samples, &root_ids, scope); + for (view, batches) in roots.into_iter().zip(outputs) { + let (key, span) = &planned[&(view.node_id as u64)]; let schema = batches[0].schema().clone(); let state_rows = batches.iter().flat_map(|b| b.to_rows().to_vec()).collect(); store.insert( @@ -427,11 +517,17 @@ fn run_data_plane_on( ); outputs_built += 1; } + } + let raw_data_reads_refreshing = RAW_DATA_READS.with(Cell::get) - raw_data_reads_before; + // Each view's state and the span of its data, by node id. + let mut built: BTreeMap = BTreeMap::new(); + for view in &views { + let (key, span) = &planned[&(view.node_id as u64)]; // One stored state serves every view with its key, so it is kept // for the largest retention any of them requires. - let stored = store.get_mut(&key).unwrap(); + let stored = store.get_mut(key).unwrap(); stored.retention_ms = stored.retention_ms.max(view.retention_ms()); - built.insert(view.node_id as u64, (stored.state.clone(), span)); + built.insert(view.node_id as u64, (stored.state.clone(), *span)); } // The query run reads the views it reaches from the queries' roots // without passing another view; those below are read only through the @@ -493,6 +589,122 @@ fn run_data_plane_on( answers_per_query, reference_answers_per_query, outputs_built, + raw_data_reads_refreshing, + } +} + +/// The sketch algorithm of the panes view `id` of `plan` builds. +fn pane_sketch(plan: &InstallablePlan, id: usize) -> SketchAlgorithm { + let node = plan.dag.nodes.iter().find(|node| node.id == id).unwrap(); + match &node.payload { + asap_executor::planner::ir::physical_export::PhysicalASAPOperatorPayload::ASAP( + ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(kind, _), + .. + }, + ) => *kind.algorithm(), + other => panic!("not a sketch pane: {other:?}"), + } +} + +/// Issue #1151's example: a p90 at rank error 0.01 over a minute and a p99 +/// at relative error 0.01 over five minutes, both every 10 s, over 1,000 +/// ingesting non-negative `latency_ms` gauges sampled every 10 s, on a +/// deployment that keeps no raw data: a KLL chain and a DDSketch chain, +/// both of 10 s panes, kept at ingestion over one `Scan`. Their panes are one refresh group, so each +/// 10 s window is read from raw data once, in one ingestion run that builds +/// both panes ending there: 30 reads for the 30 pane windows of the first +/// evaluation (36 when each pane view had its own run). Each state equals +/// the one a run of its view alone builds, and both queries answer as the +/// query-time reference. +#[tokio::test] +async fn panes_of_one_refresh_window_read_each_raw_window_once() { + let entry = |query: &str, metric: &str| { + format!( + r#"{{"query": {query:?}, "demand": {{"fixed_interval": 10000}}, + "requirements": {{"accuracy": {{"accuracy_metric": {metric:?}, "max_error": 0.01, + "failure_probability": 0.01, "accuracy_scope": "whole_result"}}}}, + "predictability": {{"predictable": {{"known_at": null}}}}}}"# + ) + }; + let workload: PlanningWorkload = serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{}, {}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 100.0}}}}, + "sources": {{"latency_ms": {{"series_count": 1000, "sample_interval": 10000, + "metric_type": "gauge", "values_non_negative": true}}}}}}}}"#, + entry("quantile_over_time(0.9, latency_ms[1m])", "rank"), + entry("quantile_over_time(0.99, latency_ms[5m])", "relative"), + )) + .unwrap(); + let capabilities = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + let plan_json = plan_json_for(workload, &capabilities).await; + let installed: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); + installed.check_compiles().unwrap(); + // One refresh group: the six 10 s KLL panes and the thirty 10 s + // DDSketch panes. + let groups = installed.ingestion_refresh_groups().unwrap(); + assert_eq!(groups.len(), 1, "{groups:#?}"); + assert_eq!( + groups[0].refresh_window, + Some(RefreshWindow { + width_ms: 10_000, + phase_ms: 0 + }) + ); + let sketches: Vec = groups[0] + .views + .iter() + .map(|view| pane_sketch(&installed, view.node_id)) + .collect(); + let count = |algorithm| sketches.iter().filter(|a| **a == algorithm).count(); + assert_eq!(count(SketchAlgorithm::Kll), 6, "{sketches:?}"); + assert_eq!(count(SketchAlgorithm::DDSketch), 30, "{sketches:?}"); + + let samples = remote_write_samples(); + let mut store = SummaryStore::new(); + let evaluated = run_data_plane(&plan_json, &samples, EVALUATION_MS, &mut store); + assert_eq!(evaluated.outputs_built, 36); + assert_eq!(evaluated.raw_data_reads_refreshing, 30); + assert_eq!( + evaluated.answers_per_query, + evaluated.reference_answers_per_query + ); + assert_eq!(evaluated.answers_per_query[0].len(), 2); + + // Each pane's state equals the one an ingestion run of its view alone + // builds. + for view in &groups[0].views { + let span = view.inputs.raw_data_windows()[0].bounds_ms(EVALUATION_MS); + let scope = Scope::Ingestion { + window_start_ms: span.0, + window_end_ms: span.1, + }; + let alone = run( + &installed.dag, + BTreeMap::new(), + &samples, + &[view.node_id as u64], + scope, + ) + .remove(0); + // `Value` has no equality; its debug form shows each state. + let rows = format!( + "{:?}", + alone + .iter() + .flat_map(|b| b.to_rows().to_vec()) + .collect::>() + ); + let stored = &store[&state_key(view, EVALUATION_MS)].state; + assert_eq!( + format!("{:?}", stored.to_rows()), + rows, + "view {}", + view.node_id + ); } } diff --git a/crates/types/src/ir/materialized_view.rs b/crates/types/src/ir/materialized_view.rs index 27e99090b..26e41bf8c 100644 --- a/crates/types/src/ir/materialized_view.rs +++ b/crates/types/src/ir/materialized_view.rs @@ -40,18 +40,19 @@ //! //! ## Refresh groups //! -//! Views refreshed in an ingestion run are refreshed by **refresh group**: -//! the views whose windows have one width `W` and end at one offset `p` -//! modulo `W` (their **refresh window**; a view with an unbounded window is -//! a group of its own). A runtime refreshes a group with one ingestion run -//! per window `(e - W, e]`, `e ≡ p (mod W)`. The run's roots are the views of -//! the group whose state for the window ending at `e` is not stored yet: in -//! steady state the newest pane of each chain, since older panes are the -//! same states, stored by earlier runs. A view of the group that a root -//! reads is computed in the run; views of other groups are materialized -//! view slots. A node several roots share runs once per run, so a `Scan` -//! below a KLL pane chain and a DDSketch pane chain of equal pane width -//! reads each pane's raw data once. +//! Views refreshed in an ingestion run are refreshed by **refresh group** +//! ([`PhysicalASAPDAG::ingestion_refresh_groups`]): the views whose windows +//! have one width `W` and end at one offset `p` modulo `W` (their +//! [`RefreshWindow`]; a view with an unbounded window is a group of its +//! own). A runtime refreshes a group with one ingestion run per window +//! `(e - W, e]`, `e ≡ p (mod W)`. The run's roots are the views of the group +//! whose state for the window ending at `e` is not stored yet: in steady +//! state the newest pane of each chain, since older panes are the same +//! states, stored by earlier runs. A view of the group that a root reads is +//! computed in the run; views of other groups are materialized view slots. +//! A node several roots share runs once per run, so a `Scan` below a KLL +//! pane chain and a DDSketch pane chain of equal pane width reads each +//! pane's raw data once. //! //! Each view's `coverage` says what its stored state is: a stable id of its //! definition (what it computes) and its selection (which rows went in, @@ -68,8 +69,8 @@ use serde::{Deserialize, Serialize}; use thiserror::Error; use super::physical_export::{ - PhysicalASAPDAG, PhysicalASAPDAGNode, PhysicalASAPDAGValidationError, PhysicalASAPNodeId, - ScanWindow, + offset_below, PhysicalASAPDAG, PhysicalASAPDAGNode, PhysicalASAPDAGValidationError, + PhysicalASAPNodeId, ScanWindow, }; use crate::ir::operator::asap::ASAPOp; use crate::ir::operator::node::{Operator, OperatorNode}; @@ -246,6 +247,49 @@ pub enum MaterializedViewRefresh { OnFirstRead, } +/// The windows an ingestion run refreshes for a view +/// ([`PhysicalASAPDAG::refresh_window`]): those `width_ms` wide that end +/// `phase_ms` (`< width_ms`) after a multiple of `width_ms`, relative to the +/// evaluation time. Views refreshed at ingestion with one refresh window +/// are refreshed together, in one ingestion run per window. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct RefreshWindow { + /// Width of each window. + pub width_ms: u64, + /// Where each window ends, modulo `width_ms`. + pub phase_ms: u64, +} + +impl RefreshWindow { + /// The refresh window of the window `(T - end_offset_ms - width_ms, T - + /// end_offset_ms]`: `(width_ms, end_offset_ms mod width_ms)`; `None` for + /// an empty window (`width_ms <= 0`). + pub fn of_window(end_offset_ms: i64, width_ms: i64) -> Option { + let phase_ms = end_offset_ms + .checked_rem_euclid(width_ms) + .filter(|_| width_ms > 0)?; + Some(RefreshWindow { + width_ms: u64::try_from(width_ms).ok()?, + phase_ms: u64::try_from(phase_ms).ok()?, + }) + } +} + +/// The views refreshed at ingestion that share one refresh window +/// ([`PhysicalASAPDAG::ingestion_refresh_groups`]). A runtime refreshes +/// them with one ingestion run per window `(e - W, e]`; each run's roots are +/// the views whose state for the window ending at `e` is not stored yet (in +/// steady state the newest pane of each chain), and a view of the group a +/// root reads is computed in the run. +#[derive(Debug, Clone, PartialEq)] +pub struct IngestionRefreshGroup { + /// The refresh window of every view of the group; `None` for a view + /// whose window is unbounded or empty, which is a group of its own. + pub refresh_window: Option, + /// The views, in dependency order. + pub views: Vec, +} + /// How far back from the evaluation time a runtime keeps a raw-data source /// for the DAG's query-time scans ([`PhysicalASAPDAG::raw_data_retention`]). #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -389,6 +433,147 @@ impl PhysicalASAPDAG { Ok(views) } + /// The views refreshed at ingestion, by refresh group: the views with + /// one [`RefreshWindow`] ([`Self::refresh_window`]) form one group, + /// which a runtime refreshes with one ingestion run per window + /// ([`IngestionRefreshGroup`]). Each group lists its views in dependency + /// order and comes after the groups whose views it reads. Errs as + /// [`Self::materialized_views`]. + pub fn ingestion_refresh_groups( + &self, + ) -> Result, InstallableDagError> { + Ok(self.ingestion_refresh_groups_of(&self.materialized_views()?)) + } + + /// [`Self::ingestion_refresh_groups`] of `views`, this DAG's + /// materialized views ([`Self::materialized_views`]), for a caller that + /// already derived them. + pub fn ingestion_refresh_groups_of( + &self, + views: &[MaterializedView], + ) -> Vec { + let views: Vec = views + .iter() + .filter(|view| view.refresh == MaterializedViewRefresh::AtIngestion) + .cloned() + .collect(); + let refresh_windows = self.refresh_windows(views.iter().map(|view| view.node_id)); + let mut groups: Vec = Vec::new(); + let mut group_by_window: HashMap = HashMap::new(); + let mut group_by_view: HashMap = HashMap::new(); + for view in views { + let refresh_window = refresh_windows[&view.node_id]; + let group = match refresh_window.and_then(|window| group_by_window.get(&window)) { + Some(&group) => group, + None => { + groups.push(IngestionRefreshGroup { + refresh_window, + views: Vec::new(), + }); + if let Some(window) = refresh_window { + group_by_window.insert(window, groups.len() - 1); + } + groups.len() - 1 + } + }; + group_by_view.insert(view.node_id, group); + groups[group].views.push(view); + } + // The other groups each group reads views of. + let reads: Vec> = groups + .iter() + .enumerate() + .map(|(group, of)| { + of.views + .iter() + .flat_map(|view| &view.inputs.materialized_views) + .filter_map(|read| group_by_view.get(read).copied()) + .filter(|&read| read != group) + .collect() + }) + .collect(); + // Each group after those it reads: a depth-first postorder. A view's + // window holds the windows of the views it reads, all ending where + // it ends, so two groups never read each other. + fn visit(group: usize, reads: &[Vec], placed: &mut [bool], order: &mut Vec) { + if std::mem::replace(&mut placed[group], true) { + return; + } + for &read in &reads[group] { + visit(read, reads, placed, order); + } + order.push(group); + } + let mut placed = vec![false; groups.len()]; + let mut order = Vec::with_capacity(groups.len()); + for group in 0..groups.len() { + visit(group, &reads, &mut placed, &mut order); + } + let mut groups: Vec> = groups.into_iter().map(Some).collect(); + order + .into_iter() + .filter_map(|group| groups[group].take()) + .collect() + } + + /// The refresh window of `node` ([`RefreshWindow::of_window`]): for a + /// `PaneChain` (a DAG with its chains unexpanded), its newest pane's, + /// `(pane_width, s mod pane_width)` with `s` the sum of the offsets of + /// the `TimeShift`s below it; for any other node, that of its window + /// ([`Self::view_windows`], which counts a chain below it as all its + /// panes). A chain and each of its expanded panes have the same one. + /// `None` when the window is unbounded or empty. O(nodes + edges). + pub fn refresh_window(&self, node: PhysicalASAPNodeId) -> Option { + self.refresh_windows([node])[&node] + } + + /// [`Self::refresh_window`] of each of `nodes`, by id, with the DAG + /// indexed once for all of them. + pub fn refresh_windows( + &self, + nodes: impl IntoIterator, + ) -> HashMap> { + let payloads = self.payloads_by_id(); + let (chains, others): (Vec<_>, Vec<_>) = nodes.into_iter().partition(|id| { + matches!( + payloads.get(id), + Some(Operator::ASAP(ASAPOp::PaneChain { .. })) + ) + }); + let mut windows: HashMap> = self + .view_windows(others) + .into_iter() + .map(|(id, window)| { + let refresh_window = match window { + ScanWindow::Within { + end_offset_ms, + width_ms, + } => RefreshWindow::of_window(end_offset_ms, width_ms), + ScanWindow::Unbounded => None, + }; + (id, refresh_window) + }) + .collect(); + for chain in chains { + let Some(Operator::ASAP(ASAPOp::PaneChain { + child, + pane_width_ms, + .. + })) = payloads.get(&chain) + else { + continue; + }; + // The newest pane ends at the offsets of the shifts below the chain. + let end_offset_ms = offset_below(*child, |id| payloads.get(&id).copied()); + let width_ms = i64::try_from(*pane_width_ms).unwrap_or(0); + windows.insert( + chain, + end_offset_ms.and_then(|offset| RefreshWindow::of_window(offset, width_ms)), + ); + } + windows + } + /// Sets `retention_ms` on every materialized view and every query-time /// `Scan` of an installed DAG (pane chains expanded), and returns the /// views ([`Self::materialized_views`]); `Err` when `deadline` passes. diff --git a/crates/types/src/ir/physical_export.rs b/crates/types/src/ir/physical_export.rs index d0d965321..b7dc9c865 100644 --- a/crates/types/src/ir/physical_export.rs +++ b/crates/types/src/ir/physical_export.rs @@ -318,6 +318,25 @@ pub fn time_range_window<'a>( } } +/// How long before the evaluation time `node` is evaluated: the sum of the +/// offsets of the `TimeShift`s from `node` down; `None` when one of them has +/// an `@` anchor or the sum overflows. +pub(crate) fn offset_below<'a>( + node: PhysicalASAPNodeId, + payload_of: impl Fn(PhysicalASAPNodeId) -> Option<&'a PhysicalASAPOperatorPayload>, +) -> Option { + let mut end_offset_ms = 0i64; + let mut below = payload_of(node); + while let Some(Operator::NonASAP(NonASAPOp::TimeShift { shift, child })) = below { + if shift.at.is_some() { + return None; + } + end_offset_ms = end_offset_ms.checked_add(shift.offset_ms)?; + below = payload_of(*child); + } + Some(end_offset_ms) +} + /// Why `PhysicalASAPDAG::with_every_node_at_query_time` failed. #[derive(Debug, Error)] pub enum ReferencePlanError { @@ -468,10 +487,11 @@ impl PhysicalASAPDAG { /// The interval of data the output of `node` summarizes, back from the /// evaluation time `T`: the union of the windows - /// ([`time_range_window`]) of every `TimeRange` below it. For an - /// expanded pane, its pane; for a view built from panes, all of them. - /// `Unbounded` when no `TimeRange` is below it or one of them is - /// unbounded (a whole SQL table, an `@` anchor). O(nodes + edges). + /// ([`time_range_window`]) of every `TimeRange` below it, and of every + /// unexpanded `PaneChain` (all its panes). For an expanded pane, its + /// pane; for a view built from panes, all of them. `Unbounded` when no + /// `TimeRange` or chain is below it or one of them is unbounded (a whole + /// SQL table, an `@` anchor). O(nodes + edges). #[cfg(any(test, feature = "test-support"))] pub fn view_window(&self, node: PhysicalASAPNodeId) -> ScanWindow { self.view_windows([node])[&node] @@ -496,12 +516,34 @@ impl PhysicalASAPDAG { if !seen.insert(id) { continue; } - if let Some(Operator::NonASAP(NonASAPOp::TimeRange { .. })) = payloads.get(&id) { - let read = time_range_window(id, |id| payloads.get(&id).copied()); - window = Some(window.map_or(read, |window| window.union(read))); - continue; - } - pending.extend(inputs.get(&id).into_iter().flatten()); + let read = match payloads.get(&id) { + Some(Operator::NonASAP(NonASAPOp::TimeRange { .. })) => { + time_range_window(id, |id| payloads.get(&id).copied()) + } + Some(Operator::ASAP(ASAPOp::PaneChain { + child, + pane_width_ms, + pane_count, + .. + })) => { + let width_ms = pane_width_ms.checked_mul(u64::from(*pane_count)); + match ( + offset_below(*child, |id| payloads.get(&id).copied()), + width_ms, + ) { + (Some(end_offset_ms), Some(width_ms)) => ScanWindow::Within { + end_offset_ms, + width_ms: i64::try_from(width_ms).unwrap_or(i64::MAX), + }, + _ => ScanWindow::Unbounded, + } + } + _ => { + pending.extend(inputs.get(&id).into_iter().flatten()); + continue; + } + }; + window = Some(window.map_or(read, |window| window.union(read))); } window.unwrap_or(ScanWindow::Unbounded) }; @@ -512,7 +554,9 @@ impl PhysicalASAPDAG { } /// Each node's operator, by id. - fn payloads_by_id(&self) -> HashMap { + pub(crate) fn payloads_by_id( + &self, + ) -> HashMap { self.nodes .iter() .map(|node| (node.id, &node.payload)) diff --git a/crates/types/tests/materialized_view.rs b/crates/types/tests/materialized_view.rs index feda9d333..500c5c9c2 100644 --- a/crates/types/tests/materialized_view.rs +++ b/crates/types/tests/materialized_view.rs @@ -6,18 +6,24 @@ use std::collections::HashMap; use std::rc::Rc; use std::time::Duration; -use asap_types::ir::materialized_view::{InstallableDagError, MaterializedViewRefresh}; +use asap_types::ir::materialized_view::{ + InstallableDagError, MaterializedViewRefresh, RefreshWindow, +}; use asap_types::ir::operator::operator_properties::{Reduction, Source}; +use asap_types::ir::operator::pane_chain::expand_pane_chains_in_workload; use asap_types::ir::operator::TimeShift; -use asap_types::ir::physical_export::{compile_physical_asap_dag, PhysicalASAPDAG}; +use asap_types::ir::physical_export::{ + compile_physical_asap_dag, compile_physical_asap_workload, PhysicalASAPDAG, +}; use asap_types::ir::scalar::ColumnRef; use asap_types::ir::schema::{ DataType, Field, FieldDataType, GroupingStrategy, Schema, SketchAlgorithm, SketchKind, SketchParams, SummaryUpdate, }; use asap_types::ir::{ - apply_materialization_timings, ASAPOp, MaterializationAssignment, NodeToTimedNodeCache, - NonASAPOp, Operator, OperatorNode, TimeRangeKind, + apply_materialization_timings, apply_materialization_timings_to_roots, ASAPOp, + MaterializationAssignment, NodeToTimedNodeCache, NonASAPOp, Operator, OperatorNode, QueryRoot, + TimeRangeKind, }; use asap_types::physical::{Materialization, Medium}; @@ -57,10 +63,9 @@ fn pane(scan: &Rc, shift_ms: i64) -> Rc { .unwrap() } -/// The panes at offsets 0 and 1 min kept at ingestion, their merge kept -/// across evaluations, exported: `(dag, merge's node id)`. -fn panes_and_kept_merge() -> (PhysicalASAPDAG, usize) { - let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { +/// `m` with columns `ts`, `value` and `job`. +fn scan_m() -> Rc { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { source: Source::TimeSeries { metric: "m".into() }, predicates: vec![], schema: Schema::with_time_index( @@ -73,7 +78,13 @@ fn panes_and_kept_merge() -> (PhysicalASAPDAG, usize) { vec![], ), })) - .unwrap(); + .unwrap() +} + +/// The panes at offsets 0 and 1 min kept at ingestion, their merge kept +/// across evaluations, exported: `(dag, merge's node id)`. +fn panes_and_kept_merge() -> (PhysicalASAPDAG, usize) { + let scan = scan_m(); let panes = [pane(&scan, 0), pane(&scan, 60_000)]; let merge = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { children: panes.to_vec(), @@ -175,3 +186,303 @@ fn the_derivation_stops_at_its_deadline() { Err(InstallableDagError::DeadlineExceeded) ); } + +/// A quantile sketch of `algorithm` (KLL k = 200 or DDSketch α = 0.01). +fn quantile_sketch(algorithm: SketchAlgorithm) -> FieldDataType { + let params = match algorithm { + SketchAlgorithm::Kll => SketchParams::Kll { + k: 200, + failure_probability: None, + }, + _ => SketchParams::DDSketch { alpha: 0.01 }, + }; + FieldDataType::Sketch(SketchKind::new(algorithm, params), Default::default()) +} + +/// `pane_count` panes of `pane_width_ms` of `family` per `job` over `input`. +fn chain( + input: &Rc, + family: FieldDataType, + pane_width_ms: u64, + pane_count: u32, +) -> Rc { + OperatorNode::new_shared(Operator::ASAP(ASAPOp::PaneChain { + child: Rc::clone(input), + family, + input: SummaryUpdate::column(ColumnRef::Named("value".into())), + reduction: Reduction::by(vec![2]), + grouping: GroupingStrategy::default(), + filter: None, + pane_width_ms, + pane_count, + })) + .unwrap() +} + +/// Three queries over one `Scan` of `m`, each merging a chain kept at +/// ingestion: a KLL of six 10 s panes, a DDSketch of thirty 10 s panes, and +/// a KLL of five 60 s panes. The DAG Stage 3 costs (chains unexpanded) and +/// the installed DAG (chains expanded into panes, retention set). +fn three_chains_over_one_scan() -> (PhysicalASAPDAG, PhysicalASAPDAG) { + let scan = scan_m(); + let chains = [ + chain(&scan, quantile_sketch(SketchAlgorithm::Kll), 10_000, 6), + chain( + &scan, + quantile_sketch(SketchAlgorithm::DDSketch), + 10_000, + 30, + ), + chain(&scan, quantile_sketch(SketchAlgorithm::Kll), 60_000, 5), + ]; + let mut assignment = MaterializationAssignment::default(); + let roots: Vec = chains + .iter() + .map(|chain| { + assignment.set(chain, Materialization::MatAtIngestion(Medium::Memory)); + QueryRoot::Operator( + OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { + children: vec![Rc::clone(chain)], + newest_panes: None, + group_by: None, + })) + .unwrap(), + ) + }) + .collect(); + let timed = apply_materialization_timings_to_roots(&roots, &assignment).unwrap(); + let unexpanded = compile_physical_asap_workload(&timed).unwrap(); + let expanded = expand_pane_chains_in_workload(&timed).unwrap(); + let mut installed = compile_physical_asap_workload(&expanded.roots).unwrap(); + installed + .set_retention(&HashMap::new(), &HashMap::new(), None) + .unwrap(); + (unexpanded, installed) +} + +/// The width and sketch of the panes `node` of `dag` builds. +fn pane_shape(dag: &PhysicalASAPDAG, node: usize) -> (u64, SketchAlgorithm) { + let payload = &dag.nodes.iter().find(|n| n.id == node).unwrap().payload; + let (family, width_ms) = match payload { + Operator::ASAP(ASAPOp::PaneChain { + family, + pane_width_ms, + .. + }) => (family, *pane_width_ms), + Operator::ASAP(ASAPOp::SummaryAgg { family, child, .. }) => { + let range = &dag.nodes.iter().find(|n| n.id == *child).unwrap().payload; + let Operator::NonASAP(NonASAPOp::TimeRange { range, .. }) = range else { + panic!("a pane reads a TimeRange") + }; + (family, range.as_millis() as u64) + } + other => panic!("not a pane: {other:?}"), + }; + let FieldDataType::Sketch(kind, _) = family else { + panic!("a sketch") + }; + (width_ms, *kind.algorithm()) +} + +/// A chain kept at ingestion has its newest pane's refresh window on the +/// DAG Stage 3 costs, and each of its expanded panes has the same one. +#[test] +fn a_chain_and_each_of_its_panes_have_one_refresh_window() { + let (unexpanded, installed) = three_chains_over_one_scan(); + let window_of = |width_ms| RefreshWindow { + width_ms, + phase_ms: 0, + }; + let chains: Vec = unexpanded + .nodes + .iter() + .filter(|node| matches!(node.payload, Operator::ASAP(ASAPOp::PaneChain { .. }))) + .map(|node| node.id) + .collect(); + assert_eq!(chains.len(), 3); + for chain in chains { + let (width_ms, _) = pane_shape(&unexpanded, chain); + assert_eq!(unexpanded.refresh_window(chain), Some(window_of(width_ms))); + } + let views = installed.materialized_views().unwrap(); + assert_eq!(views.len(), 6 + 30 + 5); + for view in &views { + let (width_ms, _) = pane_shape(&installed, view.node_id); + assert_eq!( + installed.refresh_window(view.node_id), + Some(window_of(width_ms)) + ); + } +} + +/// The 10 s KLL and 10 s DDSketch panes are one refresh group, the 60 s +/// KLL panes another; the merges, at query time, are in none. +#[test] +fn panes_of_one_width_form_one_refresh_group() { + let (_, installed) = three_chains_over_one_scan(); + let groups = installed.ingestion_refresh_groups().unwrap(); + let shapes: Vec<_> = groups + .iter() + .map(|group| { + let mut shapes: Vec<_> = group + .views + .iter() + .map(|view| pane_shape(&installed, view.node_id)) + .collect(); + shapes.sort_by_key(|(width_ms, algorithm)| (*width_ms, format!("{algorithm:?}"))); + shapes.dedup(); + (group.refresh_window, shapes) + }) + .collect(); + assert_eq!( + shapes, + [ + ( + Some(RefreshWindow { + width_ms: 10_000, + phase_ms: 0 + }), + vec![ + (10_000, SketchAlgorithm::DDSketch), + (10_000, SketchAlgorithm::Kll) + ] + ), + ( + Some(RefreshWindow { + width_ms: 60_000, + phase_ms: 0 + }), + vec![(60_000, SketchAlgorithm::Kll)] + ), + ] + ); + assert_eq!(groups[0].views.len(), 6 + 30); + assert_eq!(groups[1].views.len(), 5); +} + +/// A pane whose window ends 30 s after a multiple of its 60 s width (a +/// `TimeShift(90 s)` below its range) is refreshed on another grid than a +/// pane at no offset: another refresh group. +#[test] +fn a_shifted_pane_has_its_own_phase() { + let scan = scan_m(); + let panes = [pane(&scan, 0), pane(&scan, 90_000)]; + let mut assignment = MaterializationAssignment::default(); + for pane in &panes { + assignment.set(pane, Materialization::MatAtIngestion(Medium::Memory)); + } + let roots: Vec = panes + .iter() + .map(|pane| QueryRoot::Operator(Rc::clone(pane))) + .collect(); + let timed = apply_materialization_timings_to_roots(&roots, &assignment).unwrap(); + let mut dag = compile_physical_asap_workload(&timed).unwrap(); + dag.set_retention(&HashMap::new(), &HashMap::new(), None) + .unwrap(); + let mut windows: Vec<(u64, u64)> = dag + .ingestion_refresh_groups() + .unwrap() + .iter() + .map(|group| { + let window = group.refresh_window.unwrap(); + (window.width_ms, window.phase_ms) + }) + .collect(); + windows.sort_unstable(); + assert_eq!(windows, [(60_000, 0), (60_000, 30_000)]); +} + +/// Two views kept at ingestion over rows with no time window (`Values`) +/// have no refresh window: each is a group of its own. +#[test] +fn views_of_unbounded_windows_are_groups_of_their_own() { + let rows = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Values { + rows: vec![], + schema: scan_m().schema.clone(), + })) + .unwrap(); + let mut assignment = MaterializationAssignment::default(); + let roots: Vec = [SketchAlgorithm::Kll, SketchAlgorithm::DDSketch] + .into_iter() + .map(|algorithm| { + let build = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { + child: Rc::clone(&rows), + family: quantile_sketch(algorithm), + input: SummaryUpdate::column(ColumnRef::Named("value".into())), + reduction: Reduction::by(vec![2]), + grouping: GroupingStrategy::default(), + filter: None, + })) + .unwrap(); + assignment.set(&build, Materialization::MatAtIngestion(Medium::Memory)); + QueryRoot::Operator(build) + }) + .collect(); + let timed = apply_materialization_timings_to_roots(&roots, &assignment).unwrap(); + let mut dag = compile_physical_asap_workload(&timed).unwrap(); + dag.set_retention(&HashMap::new(), &HashMap::new(), None) + .unwrap(); + let groups = dag.ingestion_refresh_groups().unwrap(); + assert_eq!(groups.len(), 2); + for group in &groups { + assert_eq!(group.refresh_window, None); + assert_eq!(group.views.len(), 1); + } +} + +/// A rollup across jobs of a pane, both kept at ingestion, reads the pane's +/// window: one refresh group, the pane listed before the rollup that reads +/// it, so a run whose roots hold both computes the pane in the run. +#[test] +fn a_view_reading_a_view_of_its_group_follows_it() { + let scan = scan_m(); + let pane = pane(&scan, 0); + let rollup = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { + children: vec![Rc::clone(&pane)], + newest_panes: None, + group_by: Some(vec![]), + })) + .unwrap(); + let mut assignment = MaterializationAssignment::default(); + for view in [&pane, &rollup] { + assignment.set(view, Materialization::MatAtIngestion(Medium::Memory)); + } + let timed = apply_materialization_timings_to_roots( + &[QueryRoot::Operator(Rc::clone(&rollup))], + &assignment, + ) + .unwrap(); + let mut dag = compile_physical_asap_workload(&timed).unwrap(); + dag.set_retention(&HashMap::new(), &HashMap::new(), None) + .unwrap(); + let groups = dag.ingestion_refresh_groups().unwrap(); + assert_eq!(groups.len(), 1); + let [pane_view, rollup_view] = groups[0].views.as_slice() else { + panic!("{groups:#?}") + }; + assert_eq!(rollup_view.node_id, dag.roots[0]); + assert_eq!(rollup_view.inputs.materialized_views, [pane_view.node_id]); + assert_eq!( + groups[0].refresh_window, + Some(RefreshWindow { + width_ms: 60_000, + phase_ms: 0 + }) + ); +} + +/// On the DAG Stage 3 costs (chains unexpanded), a node above a chain has +/// the window of all the chain's panes: a merge of six 10 s panes, refreshed +/// at ingestion, has refresh window (60 s, 0). +#[test] +fn a_node_above_an_unexpanded_chain_has_the_window_of_its_panes() { + let (unexpanded, _) = three_chains_over_one_scan(); + let merge_of_kll_10s = unexpanded.roots[0]; + assert_eq!( + unexpanded.refresh_window(merge_of_kll_10s), + Some(RefreshWindow { + width_ms: 60_000, + phase_ms: 0 + }) + ); +} diff --git a/docs/design_docs/physical-planning-and-deployment.md b/docs/design_docs/physical-planning-and-deployment.md index 04391939f..a35a614f4 100644 --- a/docs/design_docs/physical-planning-and-deployment.md +++ b/docs/design_docs/physical-planning-and-deployment.md @@ -449,6 +449,9 @@ impl InstallablePlan { /// Every materialized view, in dependency order: its node, refresh mode /// (at ingestion, or on first read and kept), coverage, inputs and retention. pub fn materialized_views(&self) -> Result, InstallablePlanError>; + /// The views refreshed at ingestion, by refresh group (one refresh window + /// each), each group after the groups it reads. + pub fn ingestion_refresh_groups(&self) -> Result, InstallablePlanError>; /// Per raw-data `Source` read at query time: how far back raw data must be kept. pub fn raw_data_retention(&self) -> Result, InstallablePlanError>; } @@ -456,9 +459,17 @@ pub struct RawDataRetention { pub source: Source, pub lookback_ms: Option, // None = unbounded } +pub struct IngestionRefreshGroup { + pub refresh_window: Option, // None = an unbounded window, a group of its own + pub views: Vec, // in dependency order +} +pub struct RefreshWindow { + pub width_ms: u64, // width of each window + pub phase_ms: u64, // where each window ends, modulo width_ms +} ``` -Both derivations are methods on `PhysicalASAPDAG` in `asap-types`, so a +These derivations are methods on `PhysicalASAPDAG` in `asap-types`, so a runtime needs no dependency on the planner crate. A **materialized view** is a node whose output is kept across evaluations @@ -587,7 +598,8 @@ family and where the state is kept do not enter. A view with an unbounded window is a group of its own. A **refresh group** is the set of views refreshed at ingestion that have one -refresh window `(W, p)`. A runtime refreshes a refresh group with one +refresh window `(W, p)` (`PhysicalASAPDAG::refresh_window`, +`ingestion_refresh_groups`). A runtime refreshes a refresh group with one ingestion run per window `(e - W, e]`, `e ≡ p (mod W)`: - The run's roots are the views of the group whose state for the window @@ -620,7 +632,7 @@ error answered by a DDSketch chain with one-minute panes over the same `Scan`: ```text Install (once): views = plan.materialized_views()? // [KLL and DDSketch pane nodes, refresh = at ingestion, retention] - // both chains have refresh window (1m, 0): one refresh group of ten pane views + groups = plan.ingestion_refresh_groups()? // [refresh window (1m, 0): ten KLL and DDSketch pane views] refresh = compile(dag, [], [kll_pane_0, dd_pane_0]) // the newest pane of each chain // Scan → TimeRange(1m) → NativeKllBuild(k=200) // → NativeDDSketchBuild(α=0.01) @@ -725,6 +737,7 @@ operator/runtime fixtures: | --- | --- | | `kll_pane_execution::five_panes_roundtrip_and_shared_merge_runs_once` | One-minute pane refreshes from raw data → five materialized view slots → shared native merge → p50/p99; counts every sample once, checks adjacent aligned windows and instruments one merge start per run | | `kll_pane_execution::a_store_keyed_by_definition_and_time_range_answers_the_panes` | A store that finds states only by `(definition_id, time range)` answers the five-pane query: the panes share one definition and request the five ranges that tile the window | +| `runtime_install::panes_of_one_refresh_window_read_each_raw_window_once` | A KLL chain and a DDSketch chain of 10 s panes over one `Scan` are one refresh group: each pane window is read from raw data once, in one ingestion run that builds both panes; each state equals that of a run of its view alone, and both queries answer as the query-time reference | | `kll_pane_execution::panes_reject_parameters_schema_and_missing_binding` | Parameter relabelling, incompatible schemas and missing views fail explicitly | | `precompute_candidates::grouped_rate_can_be_materialized_before_or_after_grouped_sum` | Stored Rate state and stored Sum state answer like query time; summing counters before Rate does not | | `sql_to_physical::sql_filter_grouped_sum_executes_and_rebinds` | SQL text → candidate search → physical compilation → shared Scan predicates and grouped summary execution; NULL samples are ignored and fresh bindings produce new results |