From 8c25da497d4c45ca82b309b3775594b1136248ca Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 21 Oct 2026 23:44:23 +0000 Subject: [PATCH] feat(pass2): share scans of one metric that project different labels The PromQL frontend projects only the labels a query uses, and structural CSE shares only nodes with equal schemas, so scans of one metric under `quantile by (host, job)` and `quantile by (job)` were never shared. The identical-expression rule now also tries the candidate with every predicate-free scan of a metric widened to the union of the labels its scans project. Time ranges, shifts and per-series builds keep the wider rows; any other reader narrows them back with a Project that keeps its own columns by name. Only the widened candidates that share a scan are kept. Addresses #840 (questions 1 and 5). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RqhsM5XsKhxjMP6J2XXXfB --- crates/devtools/tests/stage_pipeline.rs | 116 ++++++- .../tests/joint_search_equivalence.rs | 21 +- .../src/asap_aware_cse.rs | 282 +++++++++++++++++- .../src/workload_candidates.rs | 45 ++- .../design_docs/proposals/planner-layering.md | 9 + ...andom-mixed-seed-862-long-workload-01.json | 18 +- .../random-mixed-seed-862-workload-01.json | 18 +- .../random-mixed-seed-862-x5-workload-01.json | 18 +- 8 files changed, 493 insertions(+), 34 deletions(-) diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index 55c1aac96..31515a0a3 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -1452,7 +1452,7 @@ const QUANTILE_BY_WORKLOAD: &str = r#"{"query_workload": {"language": "promql", /// Every plan that answers an instant `quantile by (G)` exactly compiles, /// so Stage 3 prices the exact baseline next to the sketches: no physical -/// plan (1,107 of them today) is rejected as `executor_cannot_compile`, and accepted +/// plan (3,357 of them today) is rejected as `executor_cannot_compile`, and accepted /// plans answer Q3 and Q4 exactly. #[test] fn exact_grouped_quantile_plans_compile_and_are_compared() { @@ -1564,3 +1564,117 @@ fn a_scan_two_queries_share_is_offered_built_once_per_evaluation() { .unwrap(); assert!((cost(separate) - 0.1530).abs() < 1e-4, "{}", cost(separate)); } + +/// #840's workload: Q1 and Q2 window `data` (1 m and 5 m), Q3 and Q4 read +/// its latest samples by (host, job) and by job. The four scans project +/// different labels, yet all read the same rows: the identical-expression +/// rule widens them to every label the queries use, so Q1–Q4 share one +/// scan (and Q3 and Q4 one time range), each query keeping its own +/// columns. Built once per evaluation, the shared scan reads the 5 m +/// window once, 200,000 samples, at one per 10 s evaluation. It does not +/// pay here (0.2064/s against 0.1700/s for separate scans): the scan +/// carries Q3's labels, and Q3 and Q4 filter the 5 m of rows for their +/// latest samples instead of scanning 15 s. Sharing Q1 and Q2's scan alone +/// is selected (0.1533/s). +#[test] +fn scans_projecting_different_labels_are_shared() { + let query = |q: &str| { + serde_json::json!({ + "query": q, + "demand": {"fixed_interval": 10000}, + "requirements": {"accuracy": {"max_error": 0.01, "failure_probability": 0.001}}, + "predictability": {"predictable": {"known_at": null}}, + }) + }; + let workload = serde_json::json!({ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + query("quantile_over_time(0.5, data[1m])"), + query("quantile_over_time(0.9, data[5m])"), + query("quantile by (host, job) (0.5, data)"), + query("quantile by (job) (0.5, data)"), + ], + }, + "data_workload": { + "arrival": {"ingesting": {"rows_per_second": 666.67}}, + "sources": {"data": {"series_count": 10000, "metric_type": "gauge", "sample_interval": 15000}}, + }, + }); + let path = std::env::temp_dir().join(format!( + "stage_pipeline_{}_union_labels_workload.json", + std::process::id() + )); + std::fs::write(&path, workload.to_string()).unwrap(); + let document = document_of(&[ + "--exhaustive", + "--max-displayed", + "100000", + "--workload", + path.to_str().unwrap(), + ]); + std::fs::remove_file(path).unwrap(); + let physical = document["stage2_physical_asap_optimization"]["candidates"] + .as_array() + .unwrap(); + let shared = physical + .iter() + .find(|c| { + c["display_name"] + == "Q1 Kll · Q2 Kll · Q3 Kll · Q4 Kll · shared Scan (Q1, Q2, Q3, Q4), \ + TimeRange (Q3, Q4) · query time, shared: Scan, TimeRange" + }) + .expect("Q1–Q4 share one scan"); + let id = shared["id"].as_str().unwrap(); + let dag: PhysicalASAPDAG = serde_json::from_value(shared["dag"].clone()).unwrap(); + let scans: Vec<_> = dag + .nodes + .iter() + .filter(|n| n.payload.kind_name() == "Scan") + .collect(); + let [scan] = &scans[..] else { + panic!("one scan: {scans:?}") + }; + let field_names: Vec<&str> = scan + .output_schema + .fields + .iter() + .map(|f| f.name.as_str()) + .collect(); + assert_eq!( + field_names, + ["ts", "value", "host", "job", "$promql_series_identity"] + ); + let cost = &document["stage3_plan_selection"]["costs"][id]; + let scan_cost = &cost["per_node"][scan.id.to_string()]; + assert_eq!(scan_cost["multiplier"], 1, "{scan_cost}"); + // 10,000 series × 20 samples in 5 m, once per 10 s evaluation. + let detail = scan_cost["detail"].as_str().unwrap(); + assert!( + detail.starts_with("scan 200000 samples; x 0.1000/s"), + "{detail}" + ); + let selection = &document["stage3_plan_selection"]; + let cost_of = |label: &str| { + let id = physical + .iter() + .find(|c| c["display_name"] == label) + .unwrap()["id"] + .as_str() + .unwrap(); + selection["costs"][id]["cost_per_second"].as_f64().unwrap() + }; + let shared = cost["cost_per_second"].as_f64().unwrap(); + let separate = cost_of("Q1 Kll · Q2 Kll · Q3 Kll · Q4 Kll"); + assert!((shared - 0.2064).abs() < 1e-3, "{shared}"); + assert!((separate - 0.1700).abs() < 1e-3, "{separate}"); + let selected = selection["selected"].as_str().unwrap(); + let selected_label = physical.iter().find(|c| c["id"] == selected).unwrap()["display_name"] + .as_str() + .unwrap(); + assert_eq!( + selected_label, + "Q1 Kll · Q2 Kll · Q3 Kll · Q4 Kll · shared Scan (Q1, Q2) · query time, shared: Scan" + ); +} diff --git a/crates/integration-tests/tests/joint_search_equivalence.rs b/crates/integration-tests/tests/joint_search_equivalence.rs index 406225272..af632784f 100644 --- a/crates/integration-tests/tests/joint_search_equivalence.rs +++ b/crates/integration-tests/tests/joint_search_equivalence.rs @@ -836,8 +836,27 @@ fn a_shared_summary_is_combined_with_a_rollup() { .unwrap() .pass2 .as_ref(); + // A build is shared by the summary-capability rule, or by the + // identical-expression rule over scans widened to one label set. + let builds_shared_by_identical_expression = + |p: &asap_logical_asap_optimization::workload_candidates::Pass2Candidate| { + p.sharing_candidates + .iter() + .filter(|c| { + matches!( + c.nodes[0].asap(), + Some( + asap_types::ir::ASAPOp::SummaryAgg { .. } + | asap_types::ir::ASAPOp::PaneChain { .. } + ) + ) + }) + .count() + }; ( - pass2.map_or(0, |p| p.shared_builds.len()), + pass2.map_or(0, |p| { + p.shared_builds.len() + builds_shared_by_identical_expression(p) + }), pass2.map_or(0, |p| p.rollup_candidates.len()), ) }; diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs index ac922b420..33a130a65 100644 --- a/crates/logical-asap-optimization/src/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs @@ -25,6 +25,11 @@ //! gives the same Pass 2 candidate as the subset with it; such a Pass 2 //! candidate is returned once, under the subset it actually shares. //! +//! Scans of one metric that project different labels read the same rows. +//! [`with_union_scan_labels`] widens them to the union of their labels, so +//! interning finds them identical; the caller also lists the Pass 2 +//! candidates of the widened roots that share a scan. +//! //! # Summary-capability rule //! //! Composition sizes each consumer's summary build alone, for its own @@ -117,7 +122,6 @@ //! reused; a sketch's heap is not, as its membership has no guarantee yet. use std::collections::HashMap; -#[cfg(test)] use std::collections::HashSet; use std::rc::Rc; @@ -125,7 +129,8 @@ use asap_types::ir::cse::share_common_sub_dags; use asap_types::ir::operator::operator_properties::{Reduction, Source}; use asap_types::ir::operator::AggIntent; use asap_types::ir::schema::{ - ExactKind, FamilyCapability, FieldDataType, GroupingStrategy, MergeInputRule, SketchKind, + ExactKind, FamilyCapability, FieldDataType, GroupingStrategy, MergeInputRule, Schema, + SketchKind, PROMQL_SERIES_IDENTITY, }; use asap_types::ir::SortKey; use asap_types::ir::{ @@ -204,6 +209,226 @@ fn pair( order.push(Rc::clone(original)); } +/// `roots` with the time-series scans of one metric made identical when +/// they project different labels, so that [`sharing_candidates`] finds +/// them: the PromQL frontend projects only the labels each query uses +/// (`ts, value, job, …` for `quantile by (job) (0.5, data)`), and +/// structural CSE shares only nodes with equal schemas. Every scan of a +/// metric without predicates projects the union of the labels its scans +/// project, in the order the walk meets them (before the series identity, +/// with the unique keys moved along). +/// The nodes that pass the labels on ([`passes_labels_on`]) keep the wider +/// rows above a widened scan; any other consumer reads them through a +/// `Project` that keeps the columns it read, by name, as +/// `relabeled_estimate` does. `None` when every metric's scans already +/// agree, when two of them give one label different types or one is closed +/// and the other not, when a node would change arity, or when a consumer +/// would have to narrow a summary state or a grouped or filtered pane chain +/// would read a `Project`. +pub fn with_union_scan_labels( + roots: &[QueryRoot], +) -> Result>, PerComputationCandidateError> { + let mut nodes = Vec::new(); + let mut seen = HashSet::new(); + for node in roots.iter().flat_map(root_operators) { + collect_children_first(node, &mut seen, &mut nodes); + } + // Each metric's union schema, `None` once two of its scans conflict. + let mut unions: Vec<(&Source, Option)> = Vec::new(); + for node in &nodes { + let Some((source, schema)) = unfiltered_series_scan(node) else { + continue; + }; + let Some(i) = unions.iter().position(|(s, _)| *s == source) else { + unions.push((source, Some(schema.clone()))); + continue; + }; + let Some(union) = unions[i].1.as_mut() else { + continue; + }; + // A scan known to hold every label and one that is not cannot share. + if union.closed != schema.closed { + unions[i].1 = None; + continue; + } + for field in &schema.fields { + match union.fields.iter().find(|f| f.name == field.name) { + Some(known) if known == field => {} + Some(_) => { + unions[i].1 = None; + break; + } + None => { + let at = union + .fields + .iter() + .position(|f| f.name == PROMQL_SERIES_IDENTITY) + .unwrap_or(union.fields.len()); + union.fields.insert(at, field.clone()); + // Unique keys (the series identity and ts) name columns + // by position. + for column in union.unique_keys.iter_mut().flatten() { + if *column >= at { + *column += 1; + } + } + } + } + } + } + let union_of = |node: &OperatorNode| { + let (source, schema) = unfiltered_series_scan(node)?; + let (_, union) = unions.iter().find(|(s, _)| *s == source)?; + union.as_ref().filter(|union| *union != schema) + }; + if !nodes.iter().any(|node| union_of(node).is_some()) { + return Ok(None); + } + let mut widened: HashMap<*const OperatorNode, Rc> = HashMap::new(); + for node in &nodes { + let out = match (union_of(node), node.non_asap()) { + (Some(union), Some(NonASAPOp::Scan { source, .. })) => { + let mut schema = union.clone(); + schema.time_index = schema.fields.iter().position(|f| f.name == "ts"); + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source: source.clone(), + predicates: Vec::new(), + schema, + }))? + } + _ if node + .children() + .iter() + .all(|c| Rc::ptr_eq(c, &widened[&Rc::as_ptr(c)])) => + { + Rc::clone(node) + } + // A node whose renamed outputs would change arity is not widened. + _ if passes_labels_on(node) => { + match node.with_new_children(|c| Rc::clone(&widened[&Rc::as_ptr(c)])) { + Ok(rebuilt) => Rc::new(rebuilt), + Err(_) => return Ok(None), + } + } + // A grouped or filtered pane chain over a `Project` would hide + // its window from plan selection. + _ if matches!(node.asap(), Some(ASAPOp::PaneChain { .. })) => { + return Ok(None); + } + _ => { + let mut children = HashMap::new(); + for child in node.children() { + let wide = &widened[&Rc::as_ptr(child)]; + let Some(narrow) = narrowed(wide, &child.schema)? else { + return Ok(None); + }; + children.insert(Rc::as_ptr(child), narrow); + } + Rc::new(node.with_new_children(|c| Rc::clone(&children[&Rc::as_ptr(c)]))?) + } + }; + widened.insert(Rc::as_ptr(node), out); + } + let mut out = Vec::with_capacity(roots.len()); + for query_root in roots { + let mut mapped = HashMap::new(); + for node in root_operators(query_root) { + let Some(narrow) = narrowed(&widened[&Rc::as_ptr(node)], &node.schema)? else { + return Ok(None); + }; + mapped.insert(Rc::as_ptr(node), narrow); + } + out.push(match query_root { + QueryRoot::Operator(node) => QueryRoot::Operator(Rc::clone(&mapped[&Rc::as_ptr(node)])), + QueryRoot::Scalar(expr) => QueryRoot::Scalar( + expr.map_operator_refs(&mut |node| Rc::clone(&mapped[&Rc::as_ptr(node)])), + ), + }); + } + recompose_guarantees(&out).map(Some) +} + +/// The source and schema of a time-series `Scan` without predicates. +fn unfiltered_series_scan(node: &OperatorNode) -> Option<(&Source, &Schema)> { + match node.non_asap() { + Some(NonASAPOp::Scan { + source: source @ Source::TimeSeries { .. }, + predicates, + schema, + }) if predicates.is_empty() => Some((source, schema)), + _ => None, + } +} + +/// Whether `node`, over a child with more labels, keeps them in its own +/// output: a time range or shift of the rows, and a per-series build or +/// pane chain without a filter (keyed by the series identity, which +/// determines every label) with its merges, estimates and finalizations. +/// The executor reads a per-series build's raw rows from its `TimeRange` +/// directly, and plan selection reads a pane chain's window from the rows +/// below it, so nothing may sit between them. +fn passes_labels_on(node: &OperatorNode) -> bool { + matches!( + node.non_asap(), + Some(NonASAPOp::TimeRange { .. } | NonASAPOp::TimeShift { .. }) + ) || matches!( + node.asap(), + Some( + ASAPOp::SummaryAgg { + reduction: Reduction::PerEntity, + filter: None, + .. + } | ASAPOp::PaneChain { + reduction: Reduction::PerEntity, + filter: None, + .. + } | ASAPOp::SummaryMerge { .. } + | ASAPOp::SummaryEstimate { .. } + | ASAPOp::FinalizeExactAccumulator { .. } + ) + ) +} + +/// `wide`, which holds every column of `schema` and perhaps more, with +/// only `schema`'s columns, by name: `wide` itself when it has exactly +/// them; `None` when `wide` is a summary state, which has no columns to +/// project. +fn narrowed( + wide: &Rc, + schema: &Schema, +) -> Result>, PerComputationCandidateError> { + use asap_types::ir::ProjectItem; + if wide.schema.fields == schema.fields { + return Ok(Some(Rc::clone(wide))); + } + if wide.asap().is_some_and(|op| op.produced_state().is_some()) { + return Ok(None); + } + let cols = schema + .fields + .iter() + .map(|field| { + let column = wide + .schema + .fields + .iter() + .position(|f| f.name == field.name && f.table == field.table) + .expect("a widened node keeps every column it had"); + ProjectItem { + alias: Some(field.name.clone()), + expr: ScalarExpr::Column(column), + } + }) + .collect(); + Ok(Some(OperatorNode::new_shared(Operator::NonASAP( + NonASAPOp::Project { + cols, + qualifier: None, + child: Rc::clone(wide), + }, + ))?)) +} + /// Every Pass 2 candidate of the identical-expression rule for the /// candidate with query roots `roots` and sharing candidates `candidates` /// ([`sharing_candidates`]): one per non-empty subset of them that gives a @@ -1945,6 +2170,59 @@ mod tests { .collect() } + /// `sum by (job) (m)` and `max by (host, job) (m)` scan `m` with other + /// labels, so nothing is shared until the scans are widened to + /// `job, host`: then the scans and the instant time ranges above them + /// are sharing candidates, and each root keeps its columns. + #[test] + fn scans_of_one_metric_with_other_labels_share_once_widened() { + let roots = roots(&["sum by (job) (m)", "max by (host, job) (m)"]); + assert!(sharing_candidates(&roots).is_empty()); + let widened = with_union_scan_labels(&roots).unwrap().unwrap(); + let scan_labels = |root: &QueryRoot| { + let QueryRoot::Operator(node) = root else { + unreachable!() + }; + OperatorNode::reachable(node) + .iter() + .find_map(|n| match n.non_asap() { + Some(NonASAPOp::Scan { schema, .. }) => Some( + schema + .fields + .iter() + .map(|f| f.name.clone()) + .collect::>(), + ), + _ => None, + }) + .unwrap() + }; + assert_eq!( + scan_labels(&widened[0]), + ["ts", "value", "job", "host", PROMQL_SERIES_IDENTITY] + ); + assert_eq!(scan_labels(&widened[0]), scan_labels(&widened[1])); + // The unique key still names the series identity and ts. + let QueryRoot::Operator(node) = &widened[0] else { + unreachable!() + }; + let scan = OperatorNode::reachable(node) + .into_iter() + .find(|n| matches!(n.non_asap(), Some(NonASAPOp::Scan { .. }))) + .unwrap(); + assert_eq!(scan.schema.unique_keys, [vec![4, 0]]); + let kinds: Vec<_> = sharing_candidates(&widened).iter().map(kinds).collect(); + assert_eq!(kinds, ["Scan", "TimeRange"]); + for (root, wide) in roots.iter().zip(&widened) { + let (QueryRoot::Operator(root), QueryRoot::Operator(wide)) = (root, wide) else { + unreachable!() + }; + assert_eq!(root.schema, wide.schema); + } + // Scans that already agree are left alone. + assert!(with_union_scan_labels(&widened).unwrap().is_none()); + } + /// Two queries over the same range selector have two sharing candidates, /// the scans and the time ranges. Sharing the time range shares the /// scan below it, so there are two Pass 2 candidates: the scan alone shared, and diff --git a/crates/logical-asap-optimization/src/workload_candidates.rs b/crates/logical-asap-optimization/src/workload_candidates.rs index febaa65a6..aec78e607 100644 --- a/crates/logical-asap-optimization/src/workload_candidates.rs +++ b/crates/logical-asap-optimization/src/workload_candidates.rs @@ -7,15 +7,15 @@ use std::rc::Rc; use asap_types::ir::flat::flatten; -use asap_types::ir::{OperatorNode, QueryRoot}; +use asap_types::ir::{NonASAPOp, OperatorNode, QueryRoot}; use asap_types::workload::QueryWorkloadEntry; use crate::asap_aware_cse::{ capability_sharing_candidates, identical_expression_pass2_candidates, pane_candidates, rollup_candidates, rollup_pass2_candidates, sharing_candidates, summary_capability_pass2_candidates, topk_reuse_candidates, topk_reuse_pass2_candidates, - window_composition_pass2_candidates, PaneCandidate, RollupCandidate, SharedPaneChain, - SharingCandidate, TopKReuseCandidate, MAX_REUSE_CANDIDATES, + window_composition_pass2_candidates, with_union_scan_labels, PaneCandidate, RollupCandidate, + SharedPaneChain, SharingCandidate, TopKReuseCandidate, MAX_REUSE_CANDIDATES, }; use crate::per_computation_candidates::{ compose_logical_candidate, PerComputationCandidateError, PerComputationCandidates, @@ -254,6 +254,7 @@ fn pass2_candidates( let mut found = Pass2List { seen: vec![flatten(roots).0], out: Vec::new(), + widened: Vec::new(), viable, }; let none = Rules::default(); @@ -267,6 +268,7 @@ fn pass2_candidates( Ok(pass2_candidates) => pass2_candidates, Err(error) => { found.out.push(Pass2::failed(&none, error)); + found.out.append(&mut found.widened); return found.out; } }; @@ -288,6 +290,7 @@ fn pass2_candidates( Err(error) => found.out.push(Pass2::failed(&rules, error)), } } + found.out.append(&mut found.widened); found.out } @@ -296,6 +299,11 @@ fn pass2_candidates( struct Pass2List<'v> { seen: Vec, out: Vec, + /// The identical-expression candidates over widened scans + /// ([`with_union_scan_labels`]), listed after `out`: a plan both lists + /// reach the same way (e.g. one per-series build shared by either rule) + /// ties, and plan selection keeps the earlier one. + widened: Vec, /// Whether a candidate, or any of its identical-expression candidates, /// can have a valid implementation; always `true` for the exhaustive /// search. @@ -333,6 +341,11 @@ impl Pass2List<'_> { /// The identical-expression Pass 2 candidates of `roots`, which the /// other rules changed as `rules` says. + /// When a metric's scans project different labels, also those of + /// `roots` with the scans widened to every label + /// ([`with_union_scan_labels`]) that share a scan: the others only read + /// wider scans than their counterparts over `roots`. They are listed + /// last (`widened`). fn identical_expression(&mut self, rules: &Rules, roots: &[QueryRoot]) { let candidates = sharing_candidates(roots); for pass2_candidate in identical_expression_pass2_candidates(roots, &candidates) { @@ -342,6 +355,31 @@ impl Pass2List<'_> { &pass2_candidate.roots, ); } + let widened = match with_union_scan_labels(roots) { + Ok(Some(widened)) => widened, + Ok(None) => return, + Err(error) => { + self.out.push(Pass2::failed(rules, error)); + return; + } + }; + let start = self.out.len(); + let candidates = sharing_candidates(&widened); + for pass2_candidate in identical_expression_pass2_candidates(&widened, &candidates) { + let shares_a_scan = pass2_candidate + .sharing_candidates + .iter() + .any(|c| matches!(c.nodes[0].non_asap(), Some(NonASAPOp::Scan { .. }))); + if shares_a_scan { + self.push( + rules, + pass2_candidate.sharing_candidates, + &pass2_candidate.roots, + ); + } + } + let widened = self.out.split_off(start); + self.widened.extend(widened); } /// The window-composition Pass 2 candidates of `roots`, which the other @@ -626,6 +664,7 @@ mod tests { let mut found = Pass2List { seen: vec![flatten(&roots).0], out: Vec::new(), + widened: Vec::new(), viable: &|_| true, }; found.reuse(&Rules::default(), &roots, &[]); diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 40273619b..4408e9293 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -552,6 +552,15 @@ or value being summarized together with its grouping. The summary input data doe not include the window; the window-composition rule compares windows separately. +A PromQL scan projects only the labels its query uses, so two scans of one +metric read the same rows with different columns (`quantile by (host, job)` +and `quantile by (job)` over `data`). The identical-expression rule treats them +as one input: the implementation widens every scan of the metric without label +matchers to the union of those labels. Time ranges, time shifts, and per-series +summaries without a filter (with the nodes that read their state) keep the +wider rows; every other consumer reads its own columns through a projection. +These candidates are kept only when they share a scan. + The examples behind these rules: * **Summary-capability rule ([Example 2](#example-2-one-summary-for-several-computations--the-summary-capability-rule-in-pass-2)).** One UnivMon over `src_ip` from diff --git a/tools/dag-viewer/examples/random-mixed-seed-862-long-workload-01.json b/tools/dag-viewer/examples/random-mixed-seed-862-long-workload-01.json index cfeb519ac..5764d917a 100644 --- a/tools/dag-viewer/examples/random-mixed-seed-862-long-workload-01.json +++ b/tools/dag-viewer/examples/random-mixed-seed-862-long-workload-01.json @@ -19655,14 +19655,14 @@ }, "display_name": "Q1 exact (nested Sum acc) · Q2 exact (nested Sum acc) · Q3 DDSketch (Sum/Count acc) · Q4 DDSketch (Sum/Count acc) · Q5 DDSketch (Sum/Count acc) · Q6 exact (Sum acc) · Q7 exact (Sum acc) · Q8 exact (Increase acc) · Q9 exact (Rate acc) · one DDSketch · Sum acc for Q1, Q2 in 90 × 20s panes, DDSketch for Q3, Q4, Q5 in 300 × 1s panes, Sum acc for Q6 in 180 × 1s panes, Sum acc for Q7 in 120 × 30s panes, Increase acc for Q8 in 90 × 20s panes, Rate acc for Q9 in 120 × 30s panes · Sum acc rolled up · shared PaneChain (Q3, Q4, Q5), SummaryMerge (Q3, Q4, Q5) · ingestion time: Sum acc ×90 panes, DDSketch ×300 panes, Sum acc ×180 panes, Sum acc ×120 panes, Increase acc ×90 panes, Rate acc ×120 panes · query time, shared: FinalizeExactAccumulator, SummaryMerge", "from_logical": "L1", - "id": "P201-m79+P1-m3" + "id": "P390-m79+P1-m3" } ], "pruned": [] }, "stage3_plan_selection": { "accepted": { - "P201-m79+P1-m3": [ + "P390-m79+P1-m3": [ { "accepted_metric": null, "accepted_regardless_of_guarantee": false, @@ -19744,7 +19744,7 @@ ] }, "costs": { - "P201-m79+P1-m3": { + "P390-m79+P1-m3": { "cost_per_second": 4.498464708606842e-6, "cost_unit": "cost_per_second", "per_node": { @@ -20200,10 +20200,10 @@ "rejected": [], "rejected_by_cause": {}, "search": { - "candidates_checked": 562, + "candidates_checked": 1103, "components": [ { - "candidates_checked": 560, + "candidates_checked": 1100, "queries": [ 0, 1, @@ -20213,10 +20213,10 @@ 5, 6 ], - "selected": "P201-m79" + "selected": "P390-m79" }, { - "candidates_checked": 1, + "candidates_checked": 2, "queries": [ 7, 8 @@ -20365,13 +20365,13 @@ "discarded": [] } ], - "raw_scan_assignments_skipped": 43363, + "raw_scan_assignments_skipped": 71257, "raw_scan_candidates_skipped": 0, "runner_up_cost_per_second": null, "searched_as_one": null }, "search_rejected": [], - "selected": "P201-m79+P1-m3", + "selected": "P390-m79+P1-m3", "stored_outputs": [ { "built_at": "ingestion_run", diff --git a/tools/dag-viewer/examples/random-mixed-seed-862-workload-01.json b/tools/dag-viewer/examples/random-mixed-seed-862-workload-01.json index 99cc0d3e3..1cd09369e 100644 --- a/tools/dag-viewer/examples/random-mixed-seed-862-workload-01.json +++ b/tools/dag-viewer/examples/random-mixed-seed-862-workload-01.json @@ -19655,14 +19655,14 @@ }, "display_name": "Q1 exact (nested Sum acc) · Q2 exact (nested Sum acc) · Q3 DDSketch (Sum/Count acc) · Q4 DDSketch (Sum/Count acc) · Q5 DDSketch (Sum/Count acc) · Q6 exact (Sum acc) · Q7 exact (Sum acc) · Q8 exact (Increase acc) · Q9 exact (Rate acc) · one DDSketch · Sum acc for Q1, Q2 in 30 × 20s panes, DDSketch for Q3, Q4, Q5 in 100 × 1s panes, Sum acc for Q6 in 60 × 1s panes, Sum acc for Q7 in 40 × 30s panes, Increase acc for Q8 in 30 × 20s panes, Rate acc for Q9 in 40 × 30s panes · Sum acc rolled up · shared PaneChain (Q3, Q4, Q5), SummaryMerge (Q3, Q4, Q5) · ingestion time: Sum acc ×30 panes, DDSketch ×100 panes, Sum acc ×60 panes, Sum acc ×40 panes, Increase acc ×30 panes, Rate acc ×40 panes · query time, shared: FinalizeExactAccumulator, SummaryMerge", "from_logical": "L1", - "id": "P201-m79+P1-m3" + "id": "P390-m79+P1-m3" } ], "pruned": [] }, "stage3_plan_selection": { "accepted": { - "P201-m79+P1-m3": [ + "P390-m79+P1-m3": [ { "accepted_metric": null, "accepted_regardless_of_guarantee": false, @@ -19744,7 +19744,7 @@ ] }, "costs": { - "P201-m79+P1-m3": { + "P390-m79+P1-m3": { "cost_per_second": 1.5553060000025173e-6, "cost_unit": "cost_per_second", "per_node": { @@ -20200,10 +20200,10 @@ "rejected": [], "rejected_by_cause": {}, "search": { - "candidates_checked": 562, + "candidates_checked": 1103, "components": [ { - "candidates_checked": 560, + "candidates_checked": 1100, "queries": [ 0, 1, @@ -20213,10 +20213,10 @@ 5, 6 ], - "selected": "P201-m79" + "selected": "P390-m79" }, { - "candidates_checked": 1, + "candidates_checked": 2, "queries": [ 7, 8 @@ -20365,13 +20365,13 @@ "discarded": [] } ], - "raw_scan_assignments_skipped": 43363, + "raw_scan_assignments_skipped": 71257, "raw_scan_candidates_skipped": 0, "runner_up_cost_per_second": null, "searched_as_one": null }, "search_rejected": [], - "selected": "P201-m79+P1-m3", + "selected": "P390-m79+P1-m3", "stored_outputs": [ { "built_at": "ingestion_run", diff --git a/tools/dag-viewer/examples/random-mixed-seed-862-x5-workload-01.json b/tools/dag-viewer/examples/random-mixed-seed-862-x5-workload-01.json index 94e9c15f9..d8fae7afe 100644 --- a/tools/dag-viewer/examples/random-mixed-seed-862-x5-workload-01.json +++ b/tools/dag-viewer/examples/random-mixed-seed-862-x5-workload-01.json @@ -19655,14 +19655,14 @@ }, "display_name": "Q1 exact (nested Sum acc) · Q2 exact (nested Sum acc) · Q3 DDSketch (Sum/Count acc) · Q4 DDSketch (Sum/Count acc) · Q5 DDSketch (Sum/Count acc) · Q6 exact (Sum acc) · Q7 exact (Sum acc) · Q8 exact (Increase acc) · Q9 exact (Rate acc) · one DDSketch · Sum acc for Q1, Q2 in 30 × 20s panes, DDSketch for Q3, Q4, Q5 in 100 × 1s panes, Sum acc for Q6 in 60 × 1s panes, Sum acc for Q7 in 40 × 30s panes, Increase acc for Q8 in 30 × 20s panes, Rate acc for Q9 in 40 × 30s panes · Sum acc rolled up · shared PaneChain (Q3, Q4, Q5), SummaryMerge (Q3, Q4, Q5) · ingestion time: Sum acc ×30 panes, DDSketch ×100 panes, Sum acc ×60 panes, Sum acc ×40 panes, Increase acc ×30 panes, Rate acc ×40 panes · query time, shared: FinalizeExactAccumulator, SummaryMerge", "from_logical": "L1", - "id": "P201-m79+P1-m3" + "id": "P390-m79+P1-m3" } ], "pruned": [] }, "stage3_plan_selection": { "accepted": { - "P201-m79+P1-m3": [ + "P390-m79+P1-m3": [ { "accepted_metric": null, "accepted_regardless_of_guarantee": false, @@ -19744,7 +19744,7 @@ ] }, "costs": { - "P201-m79+P1-m3": { + "P390-m79+P1-m3": { "cost_per_second": 7.777162034453446e-6, "cost_unit": "cost_per_second", "per_node": { @@ -20200,10 +20200,10 @@ "rejected": [], "rejected_by_cause": {}, "search": { - "candidates_checked": 562, + "candidates_checked": 1103, "components": [ { - "candidates_checked": 560, + "candidates_checked": 1100, "queries": [ 0, 1, @@ -20213,10 +20213,10 @@ 5, 6 ], - "selected": "P201-m79" + "selected": "P390-m79" }, { - "candidates_checked": 1, + "candidates_checked": 2, "queries": [ 7, 8 @@ -20365,13 +20365,13 @@ "discarded": [] } ], - "raw_scan_assignments_skipped": 43363, + "raw_scan_assignments_skipped": 71257, "raw_scan_candidates_skipped": 0, "runner_up_cost_per_second": null, "searched_as_one": null }, "search_rejected": [], - "selected": "P201-m79+P1-m3", + "selected": "P390-m79+P1-m3", "stored_outputs": [ { "built_at": "ingestion_run",