diff --git a/crates/integration-tests/tests/joint_search_equivalence.rs b/crates/integration-tests/tests/joint_search_equivalence.rs index f97afb517..4f7df147f 100644 --- a/crates/integration-tests/tests/joint_search_equivalence.rs +++ b/crates/integration-tests/tests/joint_search_equivalence.rs @@ -668,3 +668,94 @@ fn components_follow_shared_sources() { .collect(); assert_eq!(components, vec![vec![0, 2, 3], vec![1]]); } + +/// A deployment that keeps no raw data, as the MVP benchmark runs: a plan +/// is valid only when every scan runs at ingestion time. +fn no_raw_data() -> asap_logical_asap_optimization::DeploymentCapabilities { + asap_logical_asap_optimization::DeploymentCapabilities { + raw_data_retained: false, + ..asap_logical_asap_optimization::DeploymentCapabilities::default() + } +} + +/// Without raw data, the default search builds no Stage 2 assignment that +/// scans raw data at query time and enumerates no Pass 2 candidate whose +/// every assignment does: seeded random workloads, under every cost model, +/// select the exhaustive search's plan (or both find none), and some +/// assignments and candidates are actually left out. +#[test] +fn without_raw_data_random_workloads_select_the_exhaustive_plan() { + let deployment = no_raw_data(); + let models = cost_models(1); + let (mut searched, mut skipped) = (0, 0); + for seed in 1..=60u64 { + let workload = random_workload_over(seed, &["m", "n"]); + let roots = lower_promql_roots(&workload).unwrap(); + let inventory = + asap_logical_asap_optimization::per_computation_candidates::generate_per_computation_candidates( + roots.iter().cloned().enumerate().collect(), + ) + .unwrap(); + if inventory.combinations() > MAX_EXHAUSTIVE_COMBINATIONS { + continue; + } + searched += 1; + let entries: Vec<_> = workload.query_workload.entries().collect(); + let data = workload.data_workload.clone().unwrap(); + let settings = &models[seed as usize % models.len()]; + let models = PlanningModels::builtin() + .with_plan_selection(settings) + .with_deployment_capabilities(&deployment); + let run = assert_equivalent(&format!("seed {seed}"), &roots, &entries, &data, models); + if let Ok(selection) = &run.stage3_plan_selection { + skipped += selection.search.raw_scan_assignments_skipped; + } + } + assert!( + searched >= 20, + "only {searched} workloads were small enough" + ); + assert!(skipped > 0, "no assignment was left out"); +} + +/// Examples 1 and 2 and Pattern B, in a deployment that keeps no raw data, +/// under every cost model. +#[test] +fn without_raw_data_planner_layering_examples_select_the_exhaustive_plan() { + let deployment = no_raw_data(); + let example1 = example1(); + let (example2, catalog) = example2(); + let pattern_b = pattern_b(); + for (label, workload, roots) in [ + ( + "example 1", + &example1, + lower_promql_roots(&example1).unwrap(), + ), + ( + "example 2", + &example2, + lower_sql_roots(&example2, &catalog).unwrap(), + ), + ( + "pattern B", + &pattern_b, + lower_promql_roots(&pattern_b).unwrap(), + ), + ] { + let entries: Vec<_> = workload.query_workload.entries().collect(); + let data = workload.data_workload.clone().unwrap_or_default(); + for settings in cost_models(64) { + let models = PlanningModels::builtin() + .with_plan_selection(&settings) + .with_deployment_capabilities(&deployment); + assert_equivalent( + &format!("{label} without raw data under {:?}", settings.cost_model), + &roots, + &entries, + &data, + models, + ); + } + } +} diff --git a/crates/logical-asap-optimization/src/workload_candidates.rs b/crates/logical-asap-optimization/src/workload_candidates.rs index 7aba36cfc..4f7a09709 100644 --- a/crates/logical-asap-optimization/src/workload_candidates.rs +++ b/crates/logical-asap-optimization/src/workload_candidates.rs @@ -81,7 +81,7 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>( per_computation: &'a PerComputationCandidates, entries: &[QueryWorkloadEntry], ) -> impl Iterator + 'a { - stage1_logical_asap_optimization_skipping(per_computation, entries, |_| None) + stage1_logical_asap_optimization_skipping(per_computation, entries, |_| None, |_| true) } /// [`stage1_logical_asap_optimization`] without the Pass 1 choices @@ -89,10 +89,15 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>( /// `invalid_prefix(choice)` is `Some(n)` when every choice that agrees with /// `choice` on its first `n` digits is invalid (`Some(0)`: every choice), /// so enumeration skips them all at once; `None` when `choice` is valid. +/// Neither are the candidates `viable` rejects, nor the identical-expression +/// Pass 2 candidates of one it rejects: `viable(roots)` is `false` only +/// when no implementation of `roots`, nor of any of its identical-expression +/// candidates, can be valid ([`Pass2List::push`]). pub fn stage1_logical_asap_optimization_skipping<'a, Id: Clone>( per_computation: &'a PerComputationCandidates, entries: &[QueryWorkloadEntry], mut invalid_prefix: impl FnMut(&[usize]) -> Option + 'a, + viable: impl Fn(&[QueryRoot]) -> bool + 'a, ) -> impl Iterator + 'a { let evaluation_grid_ms: Vec> = entries .iter() @@ -125,6 +130,7 @@ pub fn stage1_logical_asap_optimization_skipping<'a, Id: Clone>( &this, &evaluation_grid_ms, from, + &viable, )); } position += 1; @@ -140,12 +146,13 @@ pub fn stage1_candidates_of_choice( per_computation: &PerComputationCandidates, entries: &[QueryWorkloadEntry], choice: &[usize], + viable: &dyn Fn(&[QueryRoot]) -> bool, ) -> Vec { let evaluation_grid_ms: Vec> = entries .iter() .map(QueryWorkloadEntry::evaluation_grid_ms) .collect(); - choice_candidates(per_computation, choice, &evaluation_grid_ms, 0) + choice_candidates(per_computation, choice, &evaluation_grid_ms, 0, viable) } /// [`stage1_candidates_of_choice`] with each query's evaluation grid, its @@ -155,18 +162,28 @@ fn choice_candidates( choice: &[usize], evaluation_grid_ms: &[Option], from: usize, + viable: &dyn Fn(&[QueryRoot]) -> bool, ) -> Vec { let roots = compose_logical_candidate(per_computation, choice) .map(|roots| roots.into_iter().map(|(_, root)| root).collect::>()); - let pass2 = match &roots { - Ok(roots) => pass2_candidates(roots, evaluation_grid_ms), - Err(_) => Vec::new(), + let (pass2, composed_viable) = match &roots { + Ok(roots) => { + let composed_viable = viable(roots); + ( + pass2_candidates(roots, evaluation_grid_ms, composed_viable, viable), + composed_viable, + ) + } + Err(_) => (Vec::new(), true), }; - let mut candidates = vec![Stage1Candidate { - choice: choice.to_vec(), - roots, - pass2: None, - }]; + let mut candidates = Vec::new(); + if composed_viable { + candidates.push(Stage1Candidate { + choice: choice.to_vec(), + roots, + pass2: None, + }); + } for Pass2 { rules, sharing_candidates, @@ -223,13 +240,21 @@ impl Pass2 { /// enumeration order, each distinct candidate once. `evaluation_grid_ms[q]` /// is the grid query `q`'s evaluations fall on, if it repeats at a fixed /// interval ([`QueryWorkloadEntry::evaluation_grid_ms`]). -fn pass2_candidates(roots: &[QueryRoot], evaluation_grid_ms: &[Option]) -> Vec { +fn pass2_candidates( + roots: &[QueryRoot], + evaluation_grid_ms: &[Option], + composed_viable: bool, + viable: &dyn Fn(&[QueryRoot]) -> bool, +) -> Vec { let mut found = Pass2List { seen: vec![flatten(roots).0], out: Vec::new(), + viable, }; let none = Rules::default(); - found.identical_expression(&none, roots); + if composed_viable { + found.identical_expression(&none, roots); + } found.window_composition(&none, roots, evaluation_grid_ms); found.reuse(&none, roots); let candidates = capability_sharing_candidates(roots); @@ -247,8 +272,9 @@ fn pass2_candidates(roots: &[QueryRoot], evaluation_grid_ms: &[Option]) -> }; match pass2_candidate.roots { Ok(roots) => { - found.push(&rules, Vec::new(), &roots); - found.identical_expression(&rules, &roots); + if found.push(&rules, Vec::new(), &roots) { + found.identical_expression(&rules, &roots); + } found.window_composition(&rules, &roots, evaluation_grid_ms); } // Listed as a candidate that cannot be composed; the other @@ -261,21 +287,34 @@ fn pass2_candidates(roots: &[QueryRoot], evaluation_grid_ms: &[Option]) -> /// Pass 2 candidates found so far, and the flattened DAG of every candidate /// listed (the Pass 1 candidate's included), so each is listed once. -struct Pass2List { +struct Pass2List<'v> { seen: Vec, out: Vec, + /// Whether a candidate, or any of its identical-expression candidates, + /// can have a valid implementation; always `true` for the exhaustive + /// search. + viable: &'v dyn Fn(&[QueryRoot]) -> bool, } -impl Pass2List { +impl Pass2List<'_> { + /// List the candidate with roots `roots` unless an equal one is listed + /// or it is not viable. `false` when it is not viable: neither it nor + /// its identical-expression candidates can be valid, so the caller + /// skips them. Sharing identical nodes never makes a build eligible + /// for a materialization it was not eligible for, since the shared + /// node is read by every reader of the nodes it replaces. fn push( &mut self, rules: &Rules, sharing_candidates: Vec, roots: &[QueryRoot], - ) { + ) -> bool { + if !(self.viable)(roots) { + return false; + } let flat = flatten(roots).0; if self.seen.contains(&flat) { - return; + return true; } self.seen.push(flat); self.out.push(Pass2 { @@ -283,6 +322,7 @@ impl Pass2List { sharing_candidates, roots: Ok(roots.to_vec()), }); + true } /// The identical-expression Pass 2 candidates of `roots`, which the @@ -322,8 +362,9 @@ impl Pass2List { }; match pass2_candidate.roots { Ok(roots) => { - self.push(&rules, Vec::new(), &roots); - self.identical_expression(&rules, &roots); + if self.push(&rules, Vec::new(), &roots) { + self.identical_expression(&rules, &roots); + } } Err(error) => self.out.push(Pass2::failed(&rules, error)), } @@ -367,8 +408,9 @@ impl Pass2List { .extend(pass2_candidate.rollup_candidates); match pass2_candidate.roots { Ok(roots) => { - self.push(&rules, Vec::new(), &roots); - self.identical_expression(&rules, &roots); + if self.push(&rules, Vec::new(), &roots) { + self.identical_expression(&rules, &roots); + } self.topk_reuse(&rules, &roots); } Err(error) => self.out.push(Pass2::failed(&rules, error)), @@ -394,8 +436,9 @@ impl Pass2List { .extend(pass2_candidate.topk_reuse_candidates); match pass2_candidate.roots { Ok(roots) => { - self.push(&rules, Vec::new(), &roots); - self.identical_expression(&rules, &roots); + if self.push(&rules, Vec::new(), &roots) { + self.identical_expression(&rules, &roots); + } } Err(error) => self.out.push(Pass2::failed(&rules, error)), } @@ -488,16 +531,20 @@ mod tests { let every = pass1(stage1_logical_asap_optimization(&inventory, &[]).collect()); // Every choice whose first digit is 1 is invalid. let skipped = pass1( - stage1_logical_asap_optimization_skipping(&inventory, &[], |choice| { - (choice[0] == 1).then_some(1) - }) + stage1_logical_asap_optimization_skipping( + &inventory, + &[], + |choice| (choice[0] == 1).then_some(1), + |_| true, + ) .collect(), ); let expected: Vec<_> = every.iter().filter(|c| c[0] != 1).cloned().collect(); assert_eq!(skipped, expected); assert!(skipped.len() < every.len()); assert_eq!( - stage1_logical_asap_optimization_skipping(&inventory, &[], |_| Some(0)).count(), + stage1_logical_asap_optimization_skipping(&inventory, &[], |_| Some(0), |_| true) + .count(), 0 ); } @@ -522,6 +569,7 @@ mod tests { let mut found = Pass2List { seen: vec![flatten(&roots).0], out: Vec::new(), + viable: &|_| true, }; found.reuse(&Rules::default(), &roots); let [failed] = found.out.as_slice() else { diff --git a/crates/physical-asap-optimization/src/lib.rs b/crates/physical-asap-optimization/src/lib.rs index 205603b57..487b5eb56 100644 --- a/crates/physical-asap-optimization/src/lib.rs +++ b/crates/physical-asap-optimization/src/lib.rs @@ -86,7 +86,7 @@ //! candidate alone, with the error as its reason. //! A summary needs no rewrite: `SummaryAgg` → `SummaryEstimate` already is //! build → estimate. Every node keeps the guarantee Stage 1 composed for it. -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::rc::Rc; use asap_types::ir::operator::operator_properties::Source; @@ -324,6 +324,96 @@ pub fn optimize_physical_asap( .collect() } +/// [`optimize_physical_asap`] without the materialization assignments that +/// read raw data at query time, each with its position in +/// [`optimize_physical_asap`]'s list, and how many were left out. An +/// assignment reads raw data at query time when a walk from a root reaches +/// a `Scan` without passing a summary build materialized at ingestion +/// time: that scan then runs at query time ([`reads_raw_at_query_time`]). +/// A deployment that keeps no raw data rejects every such candidate in +/// Stage 3, whatever else it chooses, so its search need not build them. +pub fn optimize_physical_asap_without_query_time_raw_scans( + from_logical: &str, + roots: &[QueryRoot], + entries: &[QueryWorkloadEntry], + data: &DataWorkload, + deployment: &DeploymentCapabilities, +) -> Result<(Vec<(usize, PhysicalCandidate)>, usize), PhysicalASAPOptimizationError> { + let mut memo = HashMap::new(); + let implemented = map_roots(roots, &mut |node| implement(node, &mut memo))?; + let mat_candidates = materialization_candidates(&implemented, entries, data, deployment); + check_panes_on_range_steps(&mat_candidates, entries)?; + let mut skipped = 0; + let mut candidates = Vec::new(); + for (position, assignment) in materialization_assignments(&mat_candidates).enumerate() { + if reads_raw_at_query_time(&implemented, &mat_candidates, &assignment) { + skipped += 1; + continue; + } + let candidate = + assigned_candidate(from_logical, &implemented, &mat_candidates, assignment)?; + candidates.push((position, candidate)); + } + Ok((candidates, skipped)) +} + +/// Whether every materialization assignment of `roots` scans raw data at +/// query time ([`reads_raw_at_query_time`]): even with every build that may +/// be materialized at ingestion time materialized there, a walk from a root +/// reaches a `Scan`. Materializing fewer builds at ingestion only leaves more +/// scans at query time. `false` when `roots` cannot be implemented (Stage 2 +/// then reports why). +pub fn every_assignment_scans_raw_at_query_time( + roots: &[QueryRoot], + entries: &[QueryWorkloadEntry], + data: &DataWorkload, + deployment: &DeploymentCapabilities, +) -> bool { + let mut memo = HashMap::new(); + let Ok(implemented) = map_roots(roots, &mut |node| implement(node, &mut memo)) else { + return false; + }; + let mat_candidates = materialization_candidates(&implemented, entries, data, deployment); + let most_at_ingestion: Vec = mat_candidates + .iter() + .map(|c| match c.at_ingestion { + Ok(()) => Materialization::MatAtIngestion(Medium::Memory), + Err(_) => Materialization::NotMaterialized, + }) + .collect(); + reads_raw_at_query_time(&implemented, &mat_candidates, &most_at_ingestion) +} + +/// Whether `assignment` of `mat_candidates` leaves a `Scan` of the +/// implemented `roots` at query time: one a walk from a root reaches +/// without passing a build materialized at ingestion time, below which +/// every node runs at ingestion time. +fn reads_raw_at_query_time( + roots: &[QueryRoot], + mat_candidates: &[MaterializationCandidate], + assignment: &[Materialization], +) -> bool { + let at_ingestion: HashSet<*const OperatorNode> = mat_candidates + .iter() + .zip(assignment) + .filter(|(_, m)| m.timing() == ExecutionTiming::IngestionTime) + .map(|(c, _)| Rc::as_ptr(&c.node)) + .collect(); + let mut seen: HashSet<*const OperatorNode> = HashSet::new(); + let mut stack: Vec<&Rc> = roots.iter().flat_map(root_operators).collect(); + while let Some(node) = stack.pop() { + let ptr = Rc::as_ptr(node); + if at_ingestion.contains(&ptr) || !seen.insert(ptr) { + continue; + } + if matches!(node.operator, Operator::NonASAP(NonASAPOp::Scan { .. })) { + return true; + } + stack.extend(node.children()); + } + false +} + /// Stage 2 for one CandidateLogicalASAPDAG and one materialization /// assignment: the physical candidate [`optimize_physical_asap`] lists for /// it. `choose(readers)` gives the option for each summary build in diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 4de87639f..bf226b47b 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -926,6 +926,15 @@ pub struct SearchReport { /// Why the workload was searched as one; `None` when its components /// were searched separately. pub searched_as_one: Option, + /// Stage 2 materialization assignments the search did not build because + /// they scan raw data at query time, in a deployment that keeps none: + /// Stage 3 would reject each of them (`RawDataNotRetained`). + pub raw_scan_assignments_skipped: u64, + /// Stage 1 candidates the search did not enumerate because every + /// materialization of theirs scans raw data at query time, in a + /// deployment that keeps none (nor their identical-expression Pass 2 + /// candidates, which inherit that). + pub raw_scan_candidates_skipped: u64, /// Under cost models 1 and 2, the cost per second of the cheapest /// valid candidate priced more than rounding (1e-12 of it) above the /// cheapest: how close the selection came to a tie. Candidates within diff --git a/crates/plan-selection/src/pass/components.rs b/crates/plan-selection/src/pass/components.rs index 4642e236c..7ad571c70 100644 --- a/crates/plan-selection/src/pass/components.rs +++ b/crates/plan-selection/src/pass/components.rs @@ -257,11 +257,16 @@ pub(crate) fn search_components( .collect(); let joint_choice = joint_choice.ok_or(SearchedAsOne::NotInJointSpace)?; let flat = flatten(&combined).0; - let (joint, pass2) = stage1_candidates_of_choice(&joint_inventory, entries, &joint_choice) - .into_iter() - .filter_map(|candidate| Some((candidate.roots.ok()?, candidate.pass2))) - .find(|(roots, _)| flatten(roots).0 == flat) - .ok_or(SearchedAsOne::NotInJointSpace)?; + let skip_raw = !models.deployment_capabilities.raw_data_retained; + let viable = |roots: &[QueryRoot]| { + super::stage1_to_stage3::viable(roots, entries, data, models, skip_raw) + }; + let (joint, pass2) = + stage1_candidates_of_choice(&joint_inventory, entries, &joint_choice, &viable) + .into_iter() + .filter_map(|candidate| Some((candidate.roots.ok()?, candidate.pass2))) + .find(|(roots, _)| flatten(roots).0 == flat) + .ok_or(SearchedAsOne::NotInJointSpace)?; let mut assignments: Vec> = runs .iter() .map(|run| { @@ -333,6 +338,10 @@ pub(crate) fn search_components( }) .collect(); selection.search.runner_up_cost_per_second = None; + selection.search.raw_scan_assignments_skipped = selections + .iter() + .map(|s| s.search.raw_scan_assignments_skipped) + .sum(); let stage3_selected_logical = selected_logical(&joint, entries, data); Ok(Stage1ToStage3Run { stage1_pass1_per_computation: per_computation diff --git a/crates/plan-selection/src/pass/query_candidates.rs b/crates/plan-selection/src/pass/query_candidates.rs index e65f1ccc8..46a582bc4 100644 --- a/crates/plan-selection/src/pass/query_candidates.rs +++ b/crates/plan-selection/src/pass/query_candidates.rs @@ -206,7 +206,7 @@ fn alone_violation( models: &PlanningModels<'_>, ) -> Option<(RejectionCause, String)> { let mut first = None; - for candidate in stage1_candidates_of_choice(alone, entries, choice) { + for candidate in stage1_candidates_of_choice(alone, entries, choice, &|_| true) { let roots = candidate.roots.ok()?; let physical = optimize_physical_asap( "alone", diff --git a/crates/plan-selection/src/pass/stage1_to_stage3.rs b/crates/plan-selection/src/pass/stage1_to_stage3.rs index 9a79a6b2c..e32253aa6 100644 --- a/crates/plan-selection/src/pass/stage1_to_stage3.rs +++ b/crates/plan-selection/src/pass/stage1_to_stage3.rs @@ -54,7 +54,8 @@ use super::query_candidates::QueryCandidates; use super::{OptimizationInput, OptimizationPass, OptimizeError, PlanOutput, QueryPlan}; use crate::PlanningModels; use crate::{ - CalibrationFit, PlanSelector, QuerySearchReport, Rejection, Selection, SelectionError, + CalibrationFit, PlanSelector, QuerySearchReport, Rejection, RejectionCause, Selection, + SelectionError, }; use asap_logical_asap_optimization::per_computation_candidates::{ generate_per_computation_candidates, PerComputationCandidateError, PerComputationCandidates, @@ -63,7 +64,10 @@ use asap_logical_asap_optimization::workload_candidates::{ stage1_logical_asap_optimization, stage1_logical_asap_optimization_skipping, Pass2Candidate, Stage1Candidate, }; -use asap_physical_asap_optimization::{optimize_physical_asap, PhysicalCandidate}; +use asap_physical_asap_optimization::{ + every_assignment_scans_raw_at_query_time, optimize_physical_asap, + optimize_physical_asap_without_query_time_raw_scans, PhysicalCandidate, +}; /// One whole-workload CandidateLogicalASAPDAG: `choice[t]` picks target /// `t`'s candidate, or its fused computation's @@ -276,13 +280,29 @@ fn search_as_one( search.keep_every_candidate, Vec::new(), calibration, + false, + &std::cell::Cell::new(0), ); return run.with_inventory(per_computation); } let queries = QueryCandidates::check(&per_computation, entries, data, models); - let stage1 = stage1_logical_asap_optimization_skipping(&per_computation, entries, |choice| { - queries.invalid_prefix(choice) - }); + // Stage 1 candidates left out because every assignment of theirs scans + // raw data at query time. + let stage1_raw_skipped = std::cell::Cell::new(0u64); + let skip_raw = + !search.keep_every_candidate && !models.deployment_capabilities.raw_data_retained; + let stage1 = stage1_logical_asap_optimization_skipping( + &per_computation, + entries, + |choice| queries.invalid_prefix(choice), + |roots| { + let viable = viable(roots, entries, data, models, skip_raw); + if !viable { + stage1_raw_skipped.set(stage1_raw_skipped.get() + 1); + } + viable + }, + ); // The calibration is fitted over every candidate, as the exhaustive // search fits it, the discarded ones included. let calibration = calibration( @@ -300,6 +320,8 @@ fn search_as_one( search.keep_every_candidate, queries.reports(), calibration, + skip_raw, + &stage1_raw_skipped, ); run.with_inventory(per_computation) } @@ -335,10 +357,33 @@ pub fn run_stage2_and_stage3( true, Vec::new(), calibration, + false, + &std::cell::Cell::new(0), ) .with_inventory(per_computation) } +/// Whether a Stage 1 candidate with roots `roots` may have a valid +/// implementation: in a deployment that keeps no raw data (`skip_raw`), +/// not when every Stage 2 assignment of it scans raw data at query time, +/// which Stage 3 rejects (`RawDataNotRetained`); identical-expression +/// sharing cannot change that ([`stage1_logical_asap_optimization_skipping`]). +pub(super) fn viable( + roots: &[QueryRoot], + entries: &[QueryWorkloadEntry], + data: &DataWorkload, + models: PlanningModels<'_>, + skip_raw: bool, +) -> bool { + !skip_raw + || !every_assignment_scans_raw_at_query_time( + roots, + entries, + data, + models.deployment_capabilities, + ) +} + /// The calibration fit of the formulas to the deployment's measured cost /// table over `stage1`'s physical candidates /// ([`PlanSelector::needs_calibration`]), numbered as Stage 2 numbers @@ -423,7 +468,13 @@ fn search_stage2_and_stage3( keep_every_candidate: bool, queries: Vec, calibration: Result, SelectionError>, + skip_query_time_raw_scans: bool, + stage1_raw_skipped: &std::cell::Cell, ) -> Searched { + // Stage 2 assignments left out because they scan raw data at query + // time, and the id of the first one. + let mut raw_scans_skipped = 0u64; + let mut first_raw_scan_skipped: Option = None; let combinations = per_computation.combinations(); let mut logical: Vec = Vec::new(); let mut physical = Vec::new(); @@ -445,10 +496,40 @@ fn search_stage2_and_stage3( let number = index + 1; let id = format!("L{number}"); if let Ok(roots) = &candidate.roots { - match optimize_physical_asap(&id, roots, entries, data, models.deployment_capabilities) - { + let implemented = match skip_query_time_raw_scans { + // Each candidate keeps its position in its Stage 1 + // candidate's full list (`P-m`); `` counts only the + // Stage 1 candidates enumerated. + true => optimize_physical_asap_without_query_time_raw_scans( + &id, + roots, + entries, + data, + models.deployment_capabilities, + ) + .map(|(candidates, skipped)| { + raw_scans_skipped += skipped as u64; + if skipped > 0 && first_raw_scan_skipped.is_none() { + let position = (0..).find(|k| !candidates.iter().any(|(p, _)| p == k)); + first_raw_scan_skipped = position.map(|k| match k { + 0 => format!("P{number}"), + k => format!("P{number}-m{k}"), + }); + } + candidates + }), + false => optimize_physical_asap( + &id, + roots, + entries, + data, + models.deployment_capabilities, + ) + .map(|candidates| candidates.into_iter().enumerate().collect()), + }; + match implemented { Ok(implemented) => { - for (k, mut p) in implemented.into_iter().enumerate() { + for (k, mut p) in implemented { p.id = match k { 0 => format!("P{number}"), k => format!("P{number}-m{k}"), @@ -495,9 +576,42 @@ fn search_stage2_and_stage3( let (selection, selected) = match finished { Ok((mut selection, selected)) => { selection.search.queries = queries; + selection.search.raw_scan_assignments_skipped = raw_scans_skipped; + selection.search.raw_scan_candidates_skipped = stage1_raw_skipped.get(); (Ok(selection), Some(selected)) } Err(SelectionError::NoValidCandidate(mut rejected)) => { + // The first candidate left out for scanning raw data at query + // time stands for them, as Stage 3 would have rejected it. + if let Some(id) = first_raw_scan_skipped.filter(|_| { + !rejected + .iter() + .any(|r| r.cause == RejectionCause::RawDataNotRetained) + }) { + rejected.push(Rejection { + id, + cause: RejectionCause::RawDataNotRetained, + reason: "scans raw data at query time, but the deployment does not retain \ + raw data" + .to_string(), + }); + } + // Stage 1 candidates left out before Stage 2 for the same reason. + let left_out = stage1_raw_skipped.get(); + if left_out > 0 + && !rejected + .iter() + .any(|r| r.cause == RejectionCause::RawDataNotRetained) + { + rejected.push(Rejection { + id: format!("{left_out} Stage 1 candidates"), + cause: RejectionCause::RawDataNotRetained, + reason: format!( + "{left_out} Stage 1 candidates scan raw data at query time in every \ + materialization, but the deployment does not retain raw data" + ), + }); + } // A query whose every own candidate was discarded leaves no // candidate to reject: its first discard says why. for (query, report) in queries.iter().enumerate() { diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index da6c1ebe3..08f8789a3 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -1254,7 +1254,7 @@ async fn plan_p99_avg(values_non_negative: bool) -> Result Result