diff --git a/crates/cost/src/execution_demand.rs b/crates/cost/src/execution_demand.rs index 82899da9d..252a53811 100644 --- a/crates/cost/src/execution_demand.rs +++ b/crates/cost/src/execution_demand.rs @@ -345,7 +345,11 @@ pub fn consumers(dag: &PhysicalASAPDAG) -> HashMap = HashMap::new(); for (id, of_node) in consumers(dag) { let entry = count.entry(id).or_default(); + // The refresh windows whose ingestion run already computes the node. + let mut runs = HashSet::new(); for consumer in of_node { + if let Consumer::Materialized(node) = consumer { + if let Some(Some(window)) = refresh_windows.get(&node) { + if !runs.insert(*window) { + continue; + } + } + } entry.0 += 1; entry.1 += match consumer { Consumer::Query(query) => query_rates[query], diff --git a/crates/cost/src/lib.rs b/crates/cost/src/lib.rs index f3e4d7d9c..3447301e3 100644 --- a/crates/cost/src/lib.rs +++ b/crates/cost/src/lib.rs @@ -79,7 +79,8 @@ pub struct ResourceUsage { pub average_memory_bytes: f64, /// The most memory live at once, assuming every evaluation overlaps: /// in-memory kept state plus each computation's working memory, once - /// per consumer that recomputes it. + /// per consumer that recomputes it, with the `MatAtIngestion` consumers + /// of one refresh window counted once. pub peak_memory_bytes: u64, /// Bytes kept on disk: state materialized on `Medium::Disk`, plus the /// raw data query-time scans read (the longest window of each source). @@ -372,8 +373,9 @@ pub struct NodeCost { /// ([`MeasuredOperationCosts::ns_per_evaluation`]) is counted on the /// node the query reads first. pub cpu_core_seconds_per_second: f64, - /// The number of consumers it is computed for (1 when materialized, - /// [`execution_demand::computations`]). + /// The number of consumers it is computed for, with the + /// `MatAtIngestion` consumers of one refresh window counted once (1 + /// when materialized, [`execution_demand::computations`]). pub multiplier: u64, /// How it was measured, e.g. "scan 1000 samples; x 1.0000/s; keeps /// ... bytes": one computation's work, the times it is computed per diff --git a/crates/cost/tests/ingestion_refresh_runs.rs b/crates/cost/tests/ingestion_refresh_runs.rs new file mode 100644 index 000000000..cf50c0d09 --- /dev/null +++ b/crates/cost/tests/ingestion_refresh_runs.rs @@ -0,0 +1,289 @@ +//! How often a `NotMaterialized` node is computed for the views refreshed +//! at ingestion above it ([`computations`]): once per refresh window, since +//! a runtime refreshes the views of one refresh window in one ingestion run +//! (issue #1151). + +use std::rc::Rc; +use std::time::Duration; + +use asap_cost::execution_demand::computations; +use asap_types::ir::operator::operator_properties::{Reduction, Source, TimeShift}; +use asap_types::ir::physical_export::{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_to_roots, ASAPOp, MaterializationAssignment, NonASAPOp, Operator, + OperatorNode, QueryRoot, TimeRangeKind, +}; +use asap_types::physical::{Materialization, Medium}; +use asap_types::workload::{QueryRecurrence, RepeatedDemand, RepetitionInterval}; + +/// `m` with columns `ts`, `value` and `job`. +fn scan() -> Rc { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("value", DataType::Float64, false), + Field::plain("job", DataType::Utf8, true), + ], + 0, + vec![], + ), + })) + .unwrap() +} + +/// 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 a quantile sketch of +/// `algorithm` per `job` over `input`. +fn chain( + input: &Rc, + algorithm: SketchAlgorithm, + pane_width_ms: u64, + pane_count: u32, +) -> Rc { + OperatorNode::new_shared(Operator::ASAP(ASAPOp::PaneChain { + child: Rc::clone(input), + family: quantile_sketch(algorithm), + input: SummaryUpdate::column(ColumnRef::Named("value".into())), + reduction: Reduction::by(vec![2]), + grouping: GroupingStrategy::default(), + filter: None, + pane_width_ms, + pane_count, + })) + .unwrap() +} + +/// One query per summary of `summaries`, merging it, with each summary +/// materialized as `materialization`: the DAG Stage 3 costs, chains +/// unexpanded. +fn dag_of(summaries: &[Rc], materialization: Materialization) -> PhysicalASAPDAG { + let mut assignment = MaterializationAssignment::default(); + let roots: Vec = summaries + .iter() + .map(|summary| { + if materialization != Materialization::NotMaterialized { + assignment.set(summary, materialization); + } + QueryRoot::Operator( + OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { + children: vec![Rc::clone(summary)], + newest_panes: None, + group_by: None, + })) + .unwrap(), + ) + }) + .collect(); + let timed = apply_materialization_timings_to_roots(&roots, &assignment).unwrap(); + compile_physical_asap_workload(&timed).unwrap() +} + +/// `(multiplier, per_second)` of `dag`'s `Scan`, each query every 10 s. +fn scan_computations(dag: &PhysicalASAPDAG) -> (u64, f64) { + computations_of(dag, |op| matches!(op, NonASAPOp::Scan { .. })) +} + +/// `(multiplier, per_second)` of the one node of `dag` whose operator +/// `is_node` picks, each query every 10 s. +fn computations_of( + dag: &PhysicalASAPDAG, + is_node: impl Fn(&NonASAPOp) -> bool, +) -> (u64, f64) { + let every_10_s = + QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(RepetitionInterval(10_000))); + let recurrence = vec![every_10_s; dag.roots.len()]; + let times = computations(dag, &recurrence, &[], 3600.0, None).unwrap(); + let node = dag + .nodes + .iter() + .find(|node| matches!(&node.payload, Operator::NonASAP(op) if is_node(op))) + .unwrap(); + times[&node.id] +} + +/// `child` evaluated `offset_ms` earlier. +fn shifted(child: &Rc, offset_ms: i64) -> Rc { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift { + shift: TimeShift { + offset_ms, + at: None, + }, + child: Rc::clone(child), + })) + .unwrap() +} + +const AT_INGESTION: Materialization = Materialization::MatAtIngestion(Medium::Memory); + +/// A KLL chain and a DDSketch chain of 10 s panes over one `Scan`, both +/// kept at ingestion, are refreshed in one ingestion run per pane: the +/// `Scan` is computed once per second of arriving data, not twice. +#[test] +fn a_scan_under_two_chains_of_one_pane_width_is_computed_once() { + let scan = scan(); + let dag = dag_of( + &[ + chain(&scan, SketchAlgorithm::Kll, 10_000, 6), + chain(&scan, SketchAlgorithm::DDSketch, 10_000, 30), + ], + AT_INGESTION, + ); + assert_eq!(scan_computations(&dag), (1, 1.0)); +} + +/// With 10 s and 60 s panes the chains need different runs (the 10 s chain +/// needs five pane ends the 60 s chain does not): the `Scan` is computed +/// once for each. +#[test] +fn a_scan_under_chains_of_two_pane_widths_is_computed_for_each() { + let scan = scan(); + let dag = dag_of( + &[ + chain(&scan, SketchAlgorithm::Kll, 10_000, 6), + chain(&scan, SketchAlgorithm::DDSketch, 60_000, 5), + ], + AT_INGESTION, + ); + assert_eq!(scan_computations(&dag), (2, 2.0)); +} + +/// Query consumers are unchanged: with both chains built at query time +/// (`NotMaterialized`), each query computes the `Scan` at its own rate. +#[test] +fn query_consumers_still_count_once_each() { + let scan = scan(); + let range = |width_ms| { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange { + range: Duration::from_millis(width_ms), + kind: TimeRangeKind::Range, + child: Rc::clone(&scan), + })) + .unwrap() + }; + let build = |input: Rc, algorithm| { + OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { + child: input, + family: quantile_sketch(algorithm), + input: SummaryUpdate::column(ColumnRef::Named("value".into())), + reduction: Reduction::by(vec![2]), + grouping: GroupingStrategy::default(), + filter: None, + })) + .unwrap() + }; + let dag = dag_of( + &[ + build(range(60_000), SketchAlgorithm::Kll), + build(range(60_000), SketchAlgorithm::DDSketch), + ], + Materialization::NotMaterialized, + ); + assert_eq!(scan_computations(&dag), (2, 0.2)); +} + +/// Two 10 s chains and a 60 s chain over one `Scan`: two refresh windows, +/// two ingestion runs. +#[test] +fn a_scan_under_chains_of_10_10_and_60_s_is_computed_for_each_width() { + let scan = scan(); + let dag = dag_of( + &[ + chain(&scan, SketchAlgorithm::Kll, 10_000, 6), + chain(&scan, SketchAlgorithm::DDSketch, 10_000, 30), + chain(&scan, SketchAlgorithm::Kll, 60_000, 5), + ], + AT_INGESTION, + ); + assert_eq!(scan_computations(&dag), (2, 2.0)); +} + +/// A 10 s chain over the `Scan` and a 10 s chain over the `Scan` shifted by +/// 5 s end their panes on different grids (phases 0 and 5 s): two runs. +#[test] +fn chains_of_one_width_and_two_phases_are_computed_apart() { + let scan = scan(); + let dag = dag_of( + &[ + chain(&scan, SketchAlgorithm::Kll, 10_000, 6), + chain( + &shifted(&scan, 5_000), + SketchAlgorithm::DDSketch, + 10_000, + 30, + ), + ], + AT_INGESTION, + ); + assert_eq!(scan_computations(&dag), (2, 2.0)); +} + +/// Two 10 s chains over one `TimeShift(10 s)` of the `Scan` (phase 0 for +/// both): the shift and the `Scan` below it run once per run. +#[test] +fn nodes_below_a_shared_intermediate_node_are_computed_once_per_run() { + let shift = shifted(&scan(), 10_000); + let dag = dag_of( + &[ + chain(&shift, SketchAlgorithm::Kll, 10_000, 6), + chain(&shift, SketchAlgorithm::DDSketch, 10_000, 30), + ], + AT_INGESTION, + ); + assert_eq!(scan_computations(&dag), (1, 1.0)); + assert_eq!( + computations_of(&dag, |op| matches!(op, NonASAPOp::TimeShift { .. })), + (1, 1.0) + ); +} + +/// Two summaries kept at ingestion over rows with no time window +/// (`Values`) have no refresh window: each counts alone. +#[test] +fn consumers_of_unbounded_windows_count_alone() { + let rows = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Values { + rows: vec![], + schema: scan().schema.clone(), + })) + .unwrap(); + let build = |algorithm| { + 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() + }; + let dag = dag_of( + &[ + build(SketchAlgorithm::Kll), + build(SketchAlgorithm::DDSketch), + ], + AT_INGESTION, + ); + assert_eq!( + computations_of(&dag, |op| matches!(op, NonASAPOp::Values { .. })), + (2, 2.0) + ); +} diff --git a/crates/integration-tests/tests/planner_layering_common/mod.rs b/crates/integration-tests/tests/planner_layering_common/mod.rs index 1308a35ff..b81b28715 100644 --- a/crates/integration-tests/tests/planner_layering_common/mod.rs +++ b/crates/integration-tests/tests/planner_layering_common/mod.rs @@ -774,7 +774,8 @@ pub fn assert_each_node_charged_per_consuming_query(run: &Run) { /// node; for a `NotMaterialized` one, the number of consumers that reach /// it. A consumer is a query (operator or scalar root) or a materialized /// node, whose walk starts at its inputs; a walk stops at materialized -/// nodes and counts each node once. +/// nodes and counts each node once. The `MatAtIngestion` consumers of one +/// refresh window share one ingestion run and count once (#1151). pub fn computations(dag: &PhysicalASAPDAG) -> BTreeMap { let materialized: BTreeSet<_> = dag .nodes @@ -782,29 +783,46 @@ pub fn computations(dag: &PhysicalASAPDAG) -> BTreeMap { .filter(|n| n.materialization != Materialization::NotMaterialized) .map(|n| n.id) .collect(); + // Each walk's start and, for a consumer refreshed at ingestion, its + // refresh window. let starts = dag .roots .iter() - .map(|&root| vec![root]) + .map(|&root| (vec![root], None)) .chain( dag.scalar_roots .iter() - .map(|(_, expr)| expr.operator_refs().into_iter().copied().collect()), + .map(|(_, expr)| (expr.operator_refs().into_iter().copied().collect(), None)), ) - .chain(materialized.iter().map(|&id| dag.producers(id))); + .chain(materialized.iter().map(|&id| { + let at_ingestion = matches!( + dag.nodes + .iter() + .find(|n| n.id == id) + .unwrap() + .materialization, + Materialization::MatAtIngestion(_) + ); + let window = at_ingestion.then(|| dag.refresh_window(id)).flatten(); + (dag.producers(id), window) + })); let mut times: BTreeMap<_, _> = dag .nodes .iter() .map(|n| (n.id, u64::from(materialized.contains(&n.id)))) .collect(); - for start in starts { + // The (node, refresh window) pairs whose ingestion run is counted. + let mut runs = HashSet::new(); + for (start, window) in starts { let mut seen = HashSet::new(); let mut stack = start; while let Some(id) = stack.pop() { if materialized.contains(&id) || !seen.insert(id) { continue; } - *times.get_mut(&id).expect("node") += 1; + if window.is_none_or(|window| runs.insert((id, window))) { + *times.get_mut(&id).expect("node") += 1; + } stack.extend(dag.producers(id)); } } diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 2b59d70a4..ad22bea8e 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -43,7 +43,12 @@ //! ([`NodeCost::multiplier`](asap_cost::NodeCost::multiplier)) at the sum of their rates. Its consumers are //! the queries whose sub-DAG reaches it and the materialized nodes above //! it, counted without looking through a materialized node: a node below -//! one is computed when that node is. Stage 2 keeps a shared node one +//! one is computed when that node is. The `MatAtIngestion` consumers of +//! one refresh window (`PhysicalASAPDAG::refresh_window`: for a pane +//! chain, its pane width) are refreshed in one ingestion run and count as +//! one consumer: a `Scan` under a KLL chain and a DDSketch chain of 10 s +//! panes is computed once per second of arriving data, under chains of +//! 10 s and 60 s panes twice. Stage 2 keeps a shared node one //! node (it copies it only for readers at two timings). A shared //! `NotMaterialized` input costs what reading it separately does (#509 //! Example 4) only when every consumer reads the same window: a query-time `Scan` is priced over the longest window diff --git a/crates/plan-selection/src/search/costing/direct_profile.rs b/crates/plan-selection/src/search/costing/direct_profile.rs index 714107eb3..56a5a090c 100644 --- a/crates/plan-selection/src/search/costing/direct_profile.rs +++ b/crates/plan-selection/src/search/costing/direct_profile.rs @@ -9,6 +9,7 @@ use crate::search::milp::direct_cost::{DirectCostProfile, NodeCostProfile, Query use crate::search::refinement::planner::CandidatePlanningContext; use asap_cost::CostModel; use asap_cost::{NodeUsage, ProducerOutput}; +use asap_types::ir::materialized_view::RefreshWindow; use asap_types::ir::physical_export::PhysicalASAPDAGNode; use asap_types::ir::properties::execution_data_state::{DataPrimitive, ExecutionDataState}; use asap_types::ir::properties::{AccuracyScope, ExecutionTiming}; @@ -415,6 +416,7 @@ fn extract_scan_class( ingestion: false, standing_cost_per_second: 0.0, raw_retention: None, + refresh_window: None, }; let profiles = cache .iter() @@ -481,6 +483,17 @@ fn profile_for_priced( }), materialized: node.materialization != Materialization::NotMaterialized, raw_retention, + // A chain maintained at ingestion refreshes its newest pane. The + // profile admits no `TimeShift` at ingestion, so that pane ends at + // the evaluation time: phase 0. + refresh_window: match &node.payload { + asap_types::ir::Operator::ASAP(ASAPOp::PaneChain { pane_width_ms, .. }) + if node.materialization == Materialization::MatAtIngestion(Medium::Memory) => + { + RefreshWindow::of_window(0, i64::try_from(*pane_width_ms).unwrap_or(0)) + } + _ => None, + }, }) } @@ -1902,6 +1915,239 @@ mod tests { } } + /// A KLL chain (p90 over a minute) and a DDSketch chain (p99 over five + /// minutes) kept at ingestion over one `Scan`, each in panes of its + /// query's interval. With equal pane widths, one ingestion run per pane + /// computes the `Scan` for both chains, and the old Stage 3 engine's + /// modeled cost equals the real evaluator's with the `Scan` computed + /// once, for one schedule (both every 10 s) and for mixed schedules + /// (every 20 s and every 40 s, 20 s panes). With 10 s and 60 s panes the + /// chains run apart and both count the `Scan` twice (#1151). + #[test] + fn chains_of_one_refresh_window_share_their_scan_in_the_direct_profile() { + // ((p90 interval, its pane width), (p99 interval, its pane width), Scan computations) + for (kll, dd, scan_computations) in [ + ((10_000, 10_000), (10_000, 10_000), 1), + ((20_000, 20_000), (40_000, 20_000), 1), + ((10_000, 10_000), (60_000, 60_000), 2), + ] { + let (modeled, evaluated, multiplier) = two_chains_over_one_scan(kll, dd); + let case = format!("p90 (every, panes) {kll:?} ms, p99 {dd:?} ms"); + assert!( + (evaluated - modeled).abs() <= 1e-9 * evaluated.max(1e-15), + "{case}: direct {modeled}, evaluator {evaluated}" + ); + assert_eq!(multiplier, scan_computations, "{case}"); + } + } + + /// [`chains_of_one_refresh_window_share_their_scan_in_the_direct_profile`] + /// with the p90 and the p99 each `(interval in ms, pane width in ms)`: + /// the direct profile's modeled cost of the selected plan, the + /// evaluator's cost of it, and the evaluator's multiplier of the `Scan`. + fn two_chains_over_one_scan(kll: (u64, u64), dd: (u64, u64)) -> (f64, f64, u64) { + use asap_logical_asap_optimization::test_support::{ + compose_logical_candidate, generate_per_computation_candidates, pane_candidates, + window_composition_pass2_candidates, + }; + use asap_types::ir::schema::SketchAlgorithm; + use asap_types::stages::HowToComputeAggregateMeasure; + use std::collections::HashMap; + use std::rc::Rc; + /// `node` and every node below it as candidates of `configured`, one + /// per node (`seen`, by address), at ingestion below a pane chain. + fn append( + node: &Rc, + configured: &mut PhysicalCandidatesWithCosts, + ingestion: bool, + seen: &mut HashMap<*const OperatorNode, PhysicalCandidateId>, + ) -> PhysicalCandidateId { + if let Some(id) = seen.get(&Rc::as_ptr(node)) { + return *id; + } + let chain = matches!(node.asap(), Some(ASAPOp::PaneChain { .. })); + let ingestion = ingestion || chain; + let mut inputs = Vec::new(); + node.operator + .map_children(|child| inputs.push(append(child, configured, ingestion, seen))); + let id = PhysicalCandidateId(configured.physical.candidates.len()); + configured.push_for_tests( + Rc::clone(node), + if ingestion { + ExecutionTiming::IngestionTime + } else { + ExecutionTiming::QueryTime + }, + if chain { + Materialization::MatAtIngestion(Medium::Memory) + } else { + Materialization::NotMaterialized + }, + 0.0, + inputs + .into_iter() + .map(|supplier| InputProducerCandidates { + candidates: vec![supplier].into(), + }) + .collect(), + ); + seen.insert(Rc::as_ptr(node), id); + id + } + let target = AccuracyTarget::max_error(0.05) + .failure_probability(0.01) + .accuracy_scope(AccuracyScope::PerValue); + // (query, interval in ms, sketch, pane width in ms) + let queries = [ + ( + "quantile_over_time(0.9, latency[1m])", + kll.0, + SketchAlgorithm::Kll, + kll.1, + ), + ( + "quantile_over_time(0.99, latency[5m])", + dd.0, + SketchAlgorithm::DDSketch, + dd.1, + ), + ]; + let originals: Vec = queries + .iter() + .map(|(query, ..)| QueryRoot::Operator(lower_promql(query, target.clone()))) + .collect(); + let generated = generate_per_computation_candidates( + &asap_logical_asap_optimization::generate_dag_of_logical_memo_groups(&originals), + ) + .unwrap(); + // Per target, the way to compute it with its query's sketch. + let choices: Vec = generated + .targets + .iter() + .zip(&queries) + .map(|(target, (_, _, algorithm, _))| { + target + .ways_to_compute + .iter() + .position(|way| { + matches!(way, HowToComputeAggregateMeasure::Summary(found) if found == algorithm) + }) + .unwrap() + }) + .collect(); + let roots = compose_logical_candidate(&generated, &choices).unwrap(); + let grid: Vec> = queries + .iter() + .map(|(_, interval, ..)| Some(*interval)) + .collect(); + // Each build in its query's pane width. + let panes: Vec<_> = pane_candidates(&roots, &grid) + .into_iter() + .filter(|pane| { + (pane.window_ms(), pane.pane_width_ms) == (60_000, kll.1) + || (pane.window_ms(), pane.pane_width_ms) == (300_000, dd.1) + }) + .collect(); + assert_eq!(panes.len(), 2, "{panes:?}"); + let roots = window_composition_pass2_candidates(&roots, &panes) + .unwrap() + .into_iter() + .find(|candidate| { + candidate.pane_candidates.len() == 2 && candidate.shared_chains.is_empty() + }) + .unwrap() + .roots + .unwrap(); + let data: DataWorkload = serde_json::from_value(serde_json::json!({ + "arrival": {"ingesting": {"rows_per_second": 100.0}}, + "sources": {"latency": {"series_count": 2, "rows_held": 30_000, + "values_non_negative": true}} + })) + .unwrap(); + let sized = crate::sized_as_pass1(&roots, &[target.clone(), target.clone()], &data); + let mut configured = PhysicalCandidatesWithCosts::empty_for_tests(); + let mut seen = HashMap::new(); + for root in &sized { + let QueryRoot::Operator(root) = root else { + panic!("operator") + }; + let id = append(root, &mut configured, false, &mut seen); + configured.dag.query_roots.push(QueryOutputCandidates { + candidates: vec![id], + }); + } + configured.sync_physical_for_tests(); + let scans: Vec = (0..configured.physical.candidates.len()) + .filter(|&id| matches!(configured.node(id).non_asap(), Some(NonASAPOp::Scan { .. }))) + .collect(); + assert_eq!(scans.len(), 1, "one shared Scan"); + let entries: Vec = queries + .iter() + .map(|(query, interval, ..)| QueryWorkloadEntry { + query: Query((*query).into()), + requirements: QueryRequirements { + accuracy: target.clone(), + max_latency_ms: None, + }, + predictability: Predictability::Predictable { known_at: None }, + recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedInterval( + RepetitionInterval(*interval as u32), + )), + range_evaluation: None, + }) + .collect(); + let mut settings = asap_cost::CostInputs::default(); + settings.resource_speed.ms_per_cpu_op *= 1000.0; + settings.resource_speed.ms_per_scanned_byte *= 1000.0; + let context = CandidatePlanningContext { + deadline: None, + refine_materialization: false, + query_roots: &originals, + realizer: &crate::search::test_realizer(&originals, &entries, &data), + entries: &entries, + data: &data, + models: crate::SelectionModels::builtin().with_cost_inputs(&settings), + calibration: asap_cost::CalibrationFit::default(), + }; + let extracted = extract(&configured, &context).unwrap().unwrap(); + assert!(extracted.supported.iter().all(|supported| *supported)); + let proposal = crate::search::milp::model::solve_direct_cost_proposal_with_metrics( + &configured.dag, + &[], + None, + &mut Default::default(), + &extracted.profile, + ) + .unwrap() + .unwrap(); + let mut rejections = + crate::search::refinement::planner::CandidateRejectionSummary::default(); + let evaluated = crate::search::refinement::planner::evaluate_configured_candidate( + &configured, + &context, + &proposal.selected, + 0, + &mut rejections, + &mut Vec::new(), + ) + .unwrap() + .unwrap_or_else(|| panic!("{rejections:?}")); + let costs = &evaluated.0.costs[&evaluated.0.selected]; + let scan_multiplier = costs + .per_node + .iter() + .find(|(_, cost)| cost.operator == asap_cost::OperatorKind::Scan) + .map(|(_, cost)| cost.multiplier) + .unwrap(); + ( + extracted + .profile + .selected_cost(&configured.dag, &proposal.selected), + costs.cost_per_second.unwrap(), + scan_multiplier, + ) + } + // Real shared KLL/DDS plans verify schedule unions and held-memory lifetime, // including two equal frequencies whose different phases cannot coalesce. #[test] diff --git a/crates/plan-selection/src/search/milp/direct_cost.rs b/crates/plan-selection/src/search/milp/direct_cost.rs index cc26f0526..9ff9253a2 100644 --- a/crates/plan-selection/src/search/milp/direct_cost.rs +++ b/crates/plan-selection/src/search/milp/direct_cost.rs @@ -4,8 +4,9 @@ //! Profiles are supplied by physical costing; this module never infers prices //! from a logical template or treats accuracy as a monetary penalty. use crate::search::candidates::graph::{QueryRootAssignment, SelectedCandidateGraph}; +use asap_types::ir::materialized_view::RefreshWindow; use good_lp::{constraint, variable, Constraint, Expression, ProblemVariables, Variable}; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; #[path = "general_schedule.rs"] mod general_schedule; @@ -43,6 +44,10 @@ pub(crate) struct NodeCostProfile { pub held_memory_cost_per_second: f64, pub materialized: bool, pub raw_retention: Option<(String, f64)>, + /// For a node maintained at ingestion, the refresh window of its + /// ingestion runs: one run computes an unmaterialized node below the + /// nodes of one refresh window once for all of them. + pub refresh_window: Option, } impl NodeCostProfile { @@ -260,7 +265,9 @@ impl DirectCostProfile { // Per unmaterialized node, its executions per second and its held // time per second: each consumer (a query, or a materialized node) // computes its own copy at its rate and holds it for its query's - // latency, or for the materialized node's longest reader's. + // latency, or for the materialized node's longest reader's; the + // nodes maintained at ingestion of one refresh window share one + // ingestion run, which computes it once for all of them. let mut execution_rates = vec![0.0; self.nodes.len()]; let mut held_seconds = vec![0.0; self.nodes.len()]; let boundaries = selected @@ -268,23 +275,34 @@ impl DirectCostProfile { .iter() .map(|root| { let query = root.query_root; - (vec![root.candidate.0], rates[query], latency[query]) + (vec![root.candidate.0], rates[query], latency[query], None) }) .chain( selected .selected_candidates .iter() .filter(|id| self.nodes[id.0].materialized) - .map(|id| (inputs[id.0].clone(), union_rate(id.0), longest(id.0))), + .map(|id| { + ( + inputs[id.0].clone(), + union_rate(id.0), + longest(id.0), + self.nodes[id.0].refresh_window, + ) + }), ); - for (mut pending, rate, lifetime) in boundaries { + // The (node, refresh window) pairs whose ingestion run is counted. + let mut runs = BTreeSet::new(); + for (mut pending, rate, lifetime, refresh_window) in boundaries { let mut seen = vec![false; self.nodes.len()]; while let Some(id) = pending.pop() { if self.nodes[id].materialized || std::mem::replace(&mut seen[id], true) { continue; } - execution_rates[id] += rate; - held_seconds[id] += rate * lifetime; + if refresh_window.is_none_or(|window| runs.insert((id, window))) { + execution_rates[id] += rate; + held_seconds[id] += rate * lifetime; + } pending.extend(&inputs[id]); } } @@ -621,14 +639,33 @@ pub(super) fn encode( roots, &check_deadline, )?; + // Each node's executions: one per boundary reaching it, except that + // the boundaries maintained at ingestion of one refresh window share one + // ingestion run, executed when any of them reaches the node. let mut boundary_count = vec![Expression::from(0.0); count]; + let mut runs: BTreeMap<(usize, RefreshWindow), Vec> = BTreeMap::new(); for flow in &flows { + let refresh_window = flow + .boundary + .checked_sub(queries) + .and_then(|consumer| profile.nodes[consumer].refresh_window); for (id, reached) in flow.reached.iter().enumerate() { - if let Some(reached) = reached { - boundary_count[id] += *reached; + match (reached, refresh_window) { + (Some(reached), Some(window)) => { + runs.entry((id, window)).or_default().push(*reached) + } + (Some(reached), None) => boundary_count[id] += *reached, + (None, _) => {} } } } + for ((id, _), reached) in runs { + let run = variables.add(variable().min(0).max(1)); + for reached in reached { + constraints.push(constraint!(run >= reached)); + } + boundary_count[id] += run; + } // Each materialized node's longest reader's latency, built on first use. let mut longest_of: BTreeMap = BTreeMap::new(); let mut longest = @@ -767,6 +804,7 @@ mod tests { held_memory_cost_per_second: 3.0, materialized: true, raw_retention: None, + refresh_window: None, }, NodeCostProfile { ingestion: false, @@ -829,6 +867,7 @@ mod tests { held_memory_cost_per_second: 4.0, materialized: true, raw_retention: None, + refresh_window: None, }, NodeCostProfile { ingestion: false, @@ -878,6 +917,7 @@ mod tests { held_memory_cost_per_second: 3.0, materialized: true, raw_retention: None, + refresh_window: None, }, NodeCostProfile { ingestion: false, diff --git a/crates/plan-selection/src/search/milp/general_schedule.rs b/crates/plan-selection/src/search/milp/general_schedule.rs index ecd9cbf5f..5906a5c49 100644 --- a/crates/plan-selection/src/search/milp/general_schedule.rs +++ b/crates/plan-selection/src/search/milp/general_schedule.rs @@ -67,7 +67,9 @@ pub(super) fn encode( } // Per node, its demands `(used, rate, holder)`: a materialized node's own // executions, and an unmaterialized node's per consumer, each held for - // its holder's lifetime. + // its holder's lifetime; the consumers maintained at ingestion of one + // refresh window share one ingestion run, one demand used when any of + // them uses the node. let mut demands: Vec> = materialized .iter() .enumerate() @@ -78,6 +80,9 @@ pub(super) fn encode( .collect() }) .collect(); + // Per (node, refresh window), the uses of the node by that window's + // consumers, and one of them, whose lifetime holds the run's result. + let mut runs: BTreeMap<(usize, RefreshWindow), (Vec, usize)> = BTreeMap::new(); for flow in boundary_flows( graph, profile, @@ -102,15 +107,29 @@ pub(super) fn encode( } else { let consumer = flow.boundary - queries; for (used, rate) in &materialized[consumer] { - demands[id].push(( - both(variables, constraints, reached, *used), - *rate, - Holder::Node(consumer), - )); + let use_of_node = both(variables, constraints, reached, *used); + match profile.nodes[consumer].refresh_window { + Some(window) => runs + .entry((id, window)) + .or_insert_with(|| (Vec::new(), consumer)) + .0 + .push(use_of_node), + None => demands[id].push((use_of_node, *rate, Holder::Node(consumer))), + } } } } } + // One ingestion run per (node, refresh window), at the ingestion rate: + // used when any consumer of the window uses the node (minimization keeps + // it at the largest use). + for ((id, _), (uses, consumer)) in runs { + let run = variables.add(variable().min(0).max(1)); + for use_of_node in uses { + constraints.push(constraint!(run >= use_of_node)); + } + demands[id].push((run, 1.0, Holder::Node(consumer))); + } let mut objective = Expression::from(profile.request_cost / divisor); let mut retained = BTreeMap::<&str, Vec<(usize, f64)>>::new(); // Each materialized node's longest reader's latency, built on first use. diff --git a/crates/plan-selection/src/search/milp/schedule_selection_tests.rs b/crates/plan-selection/src/search/milp/schedule_selection_tests.rs index 81062594e..00d54c502 100644 --- a/crates/plan-selection/src/search/milp/schedule_selection_tests.rs +++ b/crates/plan-selection/src/search/milp/schedule_selection_tests.rs @@ -98,6 +98,7 @@ fn heterogeneous_intervals_price_boundary_rates_and_longest_reader_lifetimes() { held_memory_cost_per_second: 4.0, materialized: true, raw_retention: None, + refresh_window: None, }, NodeCostProfile { ingestion: false, @@ -168,6 +169,7 @@ fn aligned_and_offset_phases_use_evaluator_union_policy() { held_memory_cost_per_second: 2.0, materialized: true, raw_retention: None, + refresh_window: None, }], }; let result = solve_direct_cost_proposal_with_metrics( @@ -222,6 +224,7 @@ fn heterogeneous_selection_matches_exhaustive_reader_assignments() { held_memory_cost_per_second: 3.0, materialized: true, raw_retention: None, + refresh_window: None, }, NodeCostProfile { per_request: false, @@ -233,6 +236,7 @@ fn heterogeneous_selection_matches_exhaustive_reader_assignments() { held_memory_cost_per_second: 1.0, materialized: true, raw_retention: None, + refresh_window: None, }, NodeCostProfile { ingestion: false, diff --git a/crates/types/src/ir/materialized_view.rs b/crates/types/src/ir/materialized_view.rs index 26e41bf8c..22937f8a0 100644 --- a/crates/types/src/ir/materialized_view.rs +++ b/crates/types/src/ir/materialized_view.rs @@ -52,7 +52,9 @@ //! 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. +//! pane's raw data once. Stage 3's cost assumes this grouping; a runtime +//! that splits a group computes the same states at a higher cost than +//! priced. //! //! 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, @@ -252,7 +254,7 @@ pub enum MaterializedViewRefresh { /// `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)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] pub struct RefreshWindow { /// Width of each window. pub width_ms: u64, diff --git a/docs/design_docs/physical-planning-and-deployment.md b/docs/design_docs/physical-planning-and-deployment.md index a35a614f4..7fced770c 100644 --- a/docs/design_docs/physical-planning-and-deployment.md +++ b/docs/design_docs/physical-planning-and-deployment.md @@ -619,9 +619,15 @@ ingestion run per window `(e - W, e]`, `e ≡ p (mod W)`: The refresh window is derived from the DAG alone (time ranges, shifts and pane widths), like the views and their retention. Sealing, lateness, threading and -the alignment of the window grid stay the deployment's. A runtime that -splits a group (for example, because its sources seal at different times) -computes the same states with more runs. +the alignment of the window grid stay the deployment's. + +Stage 3's cost assumes this grouping (`asap_cost::execution_demand::computations`): +a stateless ingestion-time node is charged once per refresh window of the +views above it, not once per view. A runtime that splits a group (for +example, because its sources seal at different times) computes the same +states at a higher cost than priced, the same status pane reuse has. A +runtime that updates states sample by sample passes each sample once +whatever the windows, so it runs cheaper than priced. ### Running example