diff --git a/crates/devtools/tests/memo_pricing.rs b/crates/devtools/tests/memo_pricing.rs index 902cf3496..784d45eec 100644 --- a/crates/devtools/tests/memo_pricing.rs +++ b/crates/devtools/tests/memo_pricing.rs @@ -67,7 +67,7 @@ struct Planned { /// `stage_pipeline` lowers each series' identity as a column /// (`lower_promql_roots`); without it (`series_identity` false, as the /// memo-enumeration tests lower), the ratio's two `quantile 0.5` have no -/// provable unique key and stay separate groups. +/// provable unique key, but they are still one group. fn planned(workload: &PlanningWorkload, series_identity: bool) -> Planned { let roots = if series_identity { lower_promql_roots(workload).unwrap() @@ -147,21 +147,6 @@ fn price_every_choice( ) } -fn assert_none_cheaper(priced: &[PricedChoice], baseline: SelectedBaseline) { - let cheaper: Vec<_> = priced - .iter() - .enumerate() - .filter(|(_, p)| p.cheaper_than_selected) - .map(|(i, p)| (format!("E{}", i + 1), p.cost_per_second)) - .collect(); - assert!( - cheaper.is_empty(), - "selected {:?}; cheaper: {cheaper:?}", - baseline.cost_per_second - ); - assert!(priced.iter().any(|p| p.valid)); -} - /// `stage_pipeline --price-enumerated 200` on `args`' workload: the listed /// choices of the enumeration's step, each priced. fn priced_document(args: &[&str]) -> Value { @@ -242,15 +227,16 @@ fn ratio_of_quantiles_no_listed_choice_is_cheaper_than_the_selected_plan() { } } -/// Lowered without series identity, the two `quantile 0.5` stay separate -/// groups: 7 × 7 × 7 = 343 DAGs, 27 of them sized for the whole budget. The -/// executor compiles none of their per-series summaries without the series -/// identity, so only the exact DAG is valid, and it is not cheaper than the -/// plan selected for the same query as `stage_pipeline` lowers it. (The -/// default search finds no plan for this lowering in its budget, so the -/// selection comes from the series-identity lowering.) +/// Lowered without series identity, the two `quantile 0.5` have no unique +/// key but return the same rows, so they are still one group: 5 × 5 = 25 +/// DAGs (two distinct approximate inputs), 9 of them sized for the whole +/// budget. The executor compiles none of their per-series summaries without +/// the series identity, so only the exact DAG is valid. Its price is not +/// compared with the plan selected for the series-identity lowering: its +/// rows carry no identity column, so it prices lower than the same exact +/// DAG with the identity (#1035). #[test] -fn ratio_of_quantiles_343_pass1_choices_none_cheaper_than_the_selected_plan() { +fn ratio_of_quantiles_25_pass1_choices_without_identity_only_exact_is_valid() { let without_identity = planned(&ratio_of_quantiles(), false); let enumeration = enumerate(&without_identity.memo, 100_000); let whole_budget = enumeration @@ -262,21 +248,14 @@ fn ratio_of_quantiles_343_pass1_choices_none_cheaper_than_the_selected_plan() { }) }) .count(); - assert_eq!((enumeration.choices.len(), whole_budget), (343, 27)); + assert_eq!((enumeration.choices.len(), whole_budget), (25, 9)); let with_identity = planned(&ratio_of_quantiles(), true); let deployment = deployment(); for enforce in [false, true] { let settings = settings(enforce); let models = models(&settings, &deployment); let baseline = selected(&with_identity, models); - let started = std::time::Instant::now(); let priced = price_every_choice(&without_identity, models, baseline); - eprintln!( - "priced {} choices in {:.2?}", - priced.len(), - started.elapsed() - ); - assert_none_cheaper(&priced, baseline); assert_eq!(priced.iter().filter(|p| p.valid).count(), 1); } } diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs index 022445c0c..063516d65 100644 --- a/crates/logical-asap-optimization/src/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs @@ -12,7 +12,8 @@ //! A *sharing candidate* ([`SharingCandidate`]) of a whole-workload //! candidate is a set of two or more of its nodes with identical input and //! computation semantics that are legal to share: the `ir::cse` interning -//! merges only ASAP nodes, scans, and nodes with a unique key. +//! merges ASAP nodes and non-ASAP nodes that return the same rows for the +//! same input rows. //! [`identical_expression_pass2_candidates`] returns one Pass 2 candidate //! per non-empty subset of the sharing candidates (Algorithm 2, line 31): //! each sharing candidate in the subset becomes one node serving every diff --git a/crates/logical-asap-optimization/src/memo.rs b/crates/logical-asap-optimization/src/memo.rs index fe57bb5a2..f87506099 100644 --- a/crates/logical-asap-optimization/src/memo.rs +++ b/crates/logical-asap-optimization/src/memo.rs @@ -235,8 +235,10 @@ fn pair( mod tests { use super::*; use crate::test_support::lower_promql; + use asap_types::ir::operator::operator_properties::BinaryOpKind; use asap_types::ir::properties::AccuracyTarget; - use asap_types::ir::ScalarExpr; + use asap_types::ir::scalar::expr_ir::ArithmeticOpKind; + use asap_types::ir::{NonASAPOp, ScalarExpr}; fn query(text: &str) -> QueryRoot { QueryRoot::Operator(lower_promql(text, AccuracyTarget::max_error(0.01))) @@ -273,6 +275,34 @@ mod tests { } } + /// A per-series quantile has no provable unique key, but it is + /// deterministic: both operands of `G2 / G2` read the same rows, so the + /// two equal sub-expressions are one group read twice. + #[test] + fn equal_keyless_deterministic_results_share_one_group() { + let roots = vec![( + 0, + QueryRoot::Operator(lower_promql( + "avg by (job) (quantile_over_time(0.5, data[5m]) / quantile_over_time(0.5, data[5m]) + quantile_over_time(0.9, data[5m]))", + AccuracyTarget::max_error(0.01).failure_probability(0.001), + )), + )]; + let memo = LogicalMemo::from_roots(&roots); + let division = memo + .groups + .iter() + .find(|group| { + matches!(group.node.non_asap(), Some(NonASAPOp::BinaryOp { operator, .. }) + if operator.kind == BinaryOpKind::Arithmetic(ArithmeticOpKind::Div)) + }) + .expect("a `G2 / G3` group"); + assert_eq!( + division.children[0], division.children[1], + "both operands of the division are one group" + ); + assert_eq!(memo.groups.len(), 7); + } + /// Scalar references are roots of their query; literals need no invented node. #[test] fn scalar_and_literal_roots_keep_workload_positions() { diff --git a/crates/logical-asap-optimization/src/memo/groups.rs b/crates/logical-asap-optimization/src/memo/groups.rs index 5fac56fc6..a0c19947f 100644 --- a/crates/logical-asap-optimization/src/memo/groups.rs +++ b/crates/logical-asap-optimization/src/memo/groups.rs @@ -4,9 +4,10 @@ //! Groups are the workload's logical DAG interned by //! [`asap_types::ir::cse::share_common_sub_dags`], the identical-expression rule's legality //! included: structurally equal nodes are one group when sharing them is -//! legal (an ASAP node, a scan, or one whose schema has a unique key). -//! Structurally equal keyless aggregates remain separate; eligible scans -//! and keyed aggregates can share a group. +//! legal: an ASAP node, or a non-ASAP node that returns the same rows for +//! the same input rows (`NonASAPOp::is_shareable_when_equal`). +//! A unique key is not needed; an unordered `LIMIT` stays one group per +//! reader. //! //! A Pass 1 target's aggregate group gets one candidate per combination of //! its measures' Pass 1 candidates: the aggregate alone composed by @@ -848,9 +849,9 @@ mod tests { /// `q0.9 / q0.5` beside a plain `q0.9`: each of the ratio's quantile /// groups offers every sketch for the whole budget and for half of it, - /// since its query has two inputs that may be approximate; the plain - /// query's quantile group (its own: its unkeyed window is not shared) - /// offers only the whole budget. A ratio of quantiles carries no + /// since its query has two inputs that may be approximate. The plain + /// query's quantile is the numerator's group: equal per-series quantiles + /// return the same rows, so they are one group with both readers. A ratio of quantiles carries no /// guarantee, so composition does not split its budget: the /// whole-budget candidate is the exhaustive search's sizing, and the /// half-budget one, larger, is never the cheaper valid one. @@ -875,12 +876,12 @@ mod tests { .collect(); assert_eq!(shares, vec![1, 2, 1, 2]); } - let plain = group_of_kind(&memo, 1, "Aggregate"); - let plain_shares: Vec = sketch_shares(&memo, plain) - .into_iter() - .map(|(_, share)| share) - .collect(); - assert_eq!(plain_shares, vec![1, 1]); + let plain = memo + .groups + .iter() + .position(|group| group.roots_of == vec![1]) + .unwrap(); + assert_eq!(plain, operand); let whole_kll = &sketch_shares(&memo, operand)[0]; assert_eq!(whole_kll.1, 1); assert_eq!( diff --git a/crates/logical-asap-optimization/src/memo/rules/summary_inputs.rs b/crates/logical-asap-optimization/src/memo/rules/summary_inputs.rs index 5b275d228..46c20c0b1 100644 --- a/crates/logical-asap-optimization/src/memo/rules/summary_inputs.rs +++ b/crates/logical-asap-optimization/src/memo/rules/summary_inputs.rs @@ -80,9 +80,9 @@ fn canonical_preprocessing( return Ok(true); } if boundaries.contains(&Rc::as_ptr(node)) { - // Logical TimeRange groups can remain distinct because their rows lack - // a unique key. The existing state-sharing rule nevertheless permits - // identical range descriptors over the same canonical source. + // Logical TimeRange groups can remain distinct when their inputs + // differ structurally. The existing state-sharing rule nevertheless + // permits identical range descriptors over the same canonical source. if let Some(group) = memo .groups .iter() diff --git a/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs b/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs index 1a2755de6..c633ae133 100644 --- a/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs +++ b/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs @@ -1,6 +1,9 @@ //! Pins the candidates the shared-producer cases generate, recorded before //! they moved onto one generation path: each workload's expanded memo //! (groups, candidates, their origins and children) hashes to a fixed value. +//! The top-k pins were re-recorded when equal keyless results became one +//! group: the readers' shared `sum_over_time` is now one group, not one per +//! reader. use super::candidates::{try_populate_pass1, CandidateMemo, CandidateOrigin}; use super::state_configurations::PaneWidthDomain; use super::state_groups::expand_concrete_summary_state_groups_in_domain; @@ -55,6 +58,23 @@ fn expanded(queries: &[(&str, f64)], step_ms: u32) -> CandidateMemo { .memo } +/// Whether one `sum_over_time` group (an aggregate directly over a window +/// of `metric`) is reached by every query in `queries`. +fn one_sum_over_time_group_for(memo: &CandidateMemo, metric: &str, queries: &[usize]) -> bool { + use asap_types::ir::operator::Source; + use asap_types::ir::NonASAPOp; + memo.groups.iter().any(|group| { + let Some(NonASAPOp::Aggregate { child, .. }) = group.node.non_asap() else { + return false; + }; + let Some(NonASAPOp::TimeRange { child: scan, .. }) = child.non_asap() else { + return false; + }; + matches!(scan.non_asap(), Some(NonASAPOp::Scan { source: Source::TimeSeries { metric: m }, .. }) if m == metric) + && queries.iter().all(|query| group.queries.contains(query)) + }) +} + fn signature(node: &Rc) -> String { serde_json::to_string(&flatten(&[QueryRoot::Operator(Rc::clone(node))]).0).unwrap() } @@ -112,7 +132,8 @@ fn top_k_reuse_candidates_are_unchanged() { .iter() .flatten() .any(|candidate| candidate.origin == CandidateOrigin::TopKReuse)); - assert_eq!(fingerprint(&memo), (313, 16940643818053118362)); + assert!(one_sum_over_time_group_for(&memo, "m_gauge", &[0, 1])); + assert_eq!(fingerprint(&memo), (69, 14825718632741491576)); } /// Quantile readers of one input at different sizes. @@ -161,5 +182,7 @@ fn several_top_k_sets_are_unchanged() { ], 60_000, ); - assert_eq!(fingerprint(&memo), (994, 9213568444657927922)); + assert!(one_sum_over_time_group_for(&memo, "m_gauge", &[0, 1, 2])); + assert!(one_sum_over_time_group_for(&memo, "other", &[3, 4])); + assert_eq!(fingerprint(&memo), (504, 3265274154981812875)); } diff --git a/crates/planner/tests/summary_sharing.rs b/crates/planner/tests/summary_sharing.rs index c08c57273..98a015f4d 100644 --- a/crates/planner/tests/summary_sharing.rs +++ b/crates/planner/tests/summary_sharing.rs @@ -497,8 +497,8 @@ async fn cross_series_p50_and_p99_share_one_producer() { assert_eq!(unique_deployments(&output), 1); } -/// An ungrouped aggregate has no unique key, so pre-ASAP CSE keeps the two -/// copies apart; their identical producers (rate, then sum) are shared here. +/// Two identical ungrouped queries: their identical producers (rate, then +/// sum) are shared. #[tokio::test] async fn identical_ungrouped_queries_share_their_producers() { let query = ("sum(rate(x[5m]))", 0.01); diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs index a6a91b768..66e716ef4 100644 --- a/crates/types/src/ir/cse.rs +++ b/crates/types/src/ir/cse.rs @@ -35,18 +35,15 @@ //! ## Legality //! //! Structural equality is necessary but not sufficient. A non-ASAP node is -//! only ever *returned* as a match for another when its output has a provable -//! unique key (`Schema::has_unique_key()`): a producer's output can only be -//! shared across consumers when its row identity is stable across reads, so -//! a `without(..)` grouping, a `Concat`/`SetOp` that -//! drops its keys, … is always inserted fresh even when it is structurally -//! identical to something already interned. A `Scan` is the exception: equal -//! scans read the same source rows, so every reader of one shared scan reads -//! the rows it would have read alone, a SQL table's keyless rows included. -//! Global scalar reductions are also shareable: their single result has stable -//! identity even though the schema does not declare the empty unique key. -//! Equal windows directly over scans and exact TopK retaining explicit series -//! identity also preserve the identity of their selected rows. +//! only ever *returned* as a match for another when every execution over the +//! same input rows returns the same output rows +//! ([`NonASAPOp::is_shareable_when_equal`]). Equal nodes are +//! interned children first, so two equal nodes read one input, and one copy +//! returns each reader the rows it would have computed alone. A unique key +//! is not needed: equal per-series quantiles without a provable key are one +//! node. An operator that picks rows by input order alone (an unordered +//! `LIMIT`) is always inserted fresh, because two copies may return +//! different rows. //! An ASAP node (summary state and its evaluations) has no such gate: equal //! operator, schema and guarantee make it shareable, exactly as post-ASAP //! sharing decided before this IR. @@ -216,33 +213,9 @@ impl InternTable { // for something else; it still occupies a fresh slot in the bucket // (harmless: later scans require legality of the new node too). let reusable = node.is_asap() - || node.schema.has_unique_key() - || matches!(node.non_asap(), Some(NonASAPOp::Scan { .. })) - // Equal windows over one scan preserve its row identities, even - // for keyless SQL tables. Consumers can share that window's rows. - || matches!(node.non_asap(), Some(NonASAPOp::TimeRange { child, .. }) - if matches!(child.non_asap(), Some(NonASAPOp::Scan { .. }))) - // Exact PromQL TopK retains each selected series' identity. The - // ranked projection can drop the timestamp part of the input key, - // but identical selections still return the same series rows. - || (node.result_kind == crate::ir::OperatorResultKind::InstantVector - && node.schema.fields.iter().any(|field| field.name == crate::ir::schema::PROMQL_SERIES_IDENTITY) - && matches!(node.non_asap(), Some(NonASAPOp::Aggregate { measures, .. }) - if matches!(measures.as_slice(), [crate::ir::operator::AggIntent::TopK { .. }]))) - || matches!(node.non_asap(), Some(NonASAPOp::Aggregate { - reduction: crate::ir::operator::Reduction::Reduce(keys), measures, .. - }) if !keys.is_without() && keys.is_empty() && measures.iter().all(|measure| matches!(measure, - crate::ir::operator::AggIntent::Count { .. } - | crate::ir::operator::AggIntent::Sum { .. } - | crate::ir::operator::AggIntent::Min { .. } - | crate::ir::operator::AggIntent::Max { .. } - | crate::ir::operator::AggIntent::Avg { .. } - | crate::ir::operator::AggIntent::StdDev { .. } - | crate::ir::operator::AggIntent::Variance { .. } - | crate::ir::operator::AggIntent::PearsonCorr { .. } - | crate::ir::operator::AggIntent::Quantile { .. } - | crate::ir::operator::AggIntent::Cardinality { .. } - ))); + || node + .non_asap() + .is_some_and(NonASAPOp::is_shareable_when_equal); let bucket = self.buckets.entry(hash).or_default(); if reusable { if let Some(existing) = bucket @@ -592,8 +565,8 @@ mod tests { "the vector read by scalar(v) is a child and must be interned" ); assert!( - !Rc::ptr_eq(&ra, &rb), - "the scalar bridge itself has no unique key and stays separate" + Rc::ptr_eq(&ra, &rb), + "the scalar bridge has no unique key but returns the same rows" ); } @@ -732,22 +705,69 @@ mod tests { ); } + /// `DISTINCT` (every column) returns the same rows each time; a `Dedup` + /// on some columns keeps an unspecified row per key. #[test] - fn dedup_gates_sharing_the_same_as_aggregate() { - // `Dedup { cols }` adds `cols` as a unique key, so two identical - // `Dedup`s merge even though their keyless `Scan`s could not. - let dedup = || { + fn dedup_shares_only_on_every_column() { + let dedup = |cols: Vec| { node(NonASAPOp::Dedup { - cols: vec![1], + cols, child: scan(), }) }; - let (ra, rb) = two_roots(dedup(), dedup()); + let (ra, rb) = two_roots(dedup(vec![]), dedup(vec![])); assert!(Rc::ptr_eq(&ra, &rb)); + let (ra, rb) = two_roots(dedup(vec![1]), dedup(vec![1])); + assert!(!Rc::ptr_eq(&ra, &rb)); } + /// `LIMIT 10` without an order may return any 10 rows, so two copies may + /// differ; one shared copy would make both readers see the same rows. #[test] - fn group_keys_gate_still_prevented_when_partition_by_without_used() { + fn unordered_limits_are_not_shared() { + let limit = || { + node(NonASAPOp::Limit { + n: Some(10), + offset: 0, + partition_by: GroupKeys::default(), + child: scan(), + }) + }; + let (ra, rb) = two_roots(limit(), limit()); + assert!(!Rc::ptr_eq(&ra, &rb)); + } + + /// A `LIMIT` over a `Sort` with the same partitions keeps the first rows + /// of a fixed order; over a `Sort` partitioned differently, which rows + /// come first still depends on input order. + #[test] + fn limits_over_a_sort_with_their_partitions_are_shared() { + let limit = |sort_partitions: GroupKeys| { + node(NonASAPOp::Limit { + n: Some(10), + offset: 0, + partition_by: GroupKeys::default(), + child: node(NonASAPOp::Sort { + keys: vec![crate::ir::scalar::SortKey { + expr: ScalarExpr::column(2), + ascending: false, + nulls_first: false, + }], + partition_by: sort_partitions, + child: scan(), + }), + }) + }; + let (ra, rb) = two_roots(limit(GroupKeys::default()), limit(GroupKeys::default())); + assert!(Rc::ptr_eq(&ra, &rb)); + let (ra, rb) = two_roots(limit(GroupKeys::by(vec![1])), limit(GroupKeys::by(vec![1]))); + assert!(!Rc::ptr_eq(&ra, &rb)); + } + + /// `count without (service)` has no provable unique key, but it computes + /// the same rows from the same input, so equal copies are one node. + #[test] + fn without_grouping_is_shared() { let without_agg = || { node(NonASAPOp::Aggregate { reduction: Reduction::Reduce(GroupKeys::without(vec![0])), @@ -763,7 +783,7 @@ mod tests { let a = without_agg(); assert!(!a.schema.has_unique_key()); let (ra, rb) = two_roots(a, without_agg()); - assert!(!Rc::ptr_eq(&ra, &rb)); + assert!(Rc::ptr_eq(&ra, &rb)); } #[test] @@ -788,14 +808,20 @@ mod tests { let (done, completion) = std::sync::mpsc::channel(); let worker = std::thread::spawn(move || { fn keyed_diamond() -> Rc { - // BinaryOp over a keyless scan has no unique key at any level, - // so none of the 24 levels is shareable; the `Dedup` on top is. + // An unordered `Limit` at each of the 24 levels is never + // shared, so comparing the `Dedup` on top walks the diamond. let mut current = scan(); for _ in 0..24 { - current = compare(Rc::clone(¤t), current); + let limited = node(NonASAPOp::Limit { + n: Some(10), + offset: 0, + partition_by: GroupKeys::default(), + child: current, + }); + current = compare(Rc::clone(&limited), limited); } node(NonASAPOp::Dedup { - cols: vec![1], + cols: vec![], child: current, }) } diff --git a/crates/types/src/ir/operator/non_asap.rs b/crates/types/src/ir/operator/non_asap.rs index d945ea841..8d3b0cc6b 100644 --- a/crates/types/src/ir/operator/non_asap.rs +++ b/crates/types/src/ir/operator/non_asap.rs @@ -10,7 +10,8 @@ use super::node::{OperatorNode, OperatorResultKind}; use crate::ir::operator::agg_intent::AggIntent; use crate::ir::operator::operator_properties::{ BinaryOpKind, ConcatDiscriminatorKey, GroupKeys, InfoMatcher, JoinKind, Reduction, - RelationalSetOpKind, SampleKind, Source, TimeShift, VectorMatch, WindowFrame, WindowFuncKind, + RelationalSetOpKind, SampleKind, Source, TimeShift, VectorMatch, WindowFrame, WindowFrameUnits, + WindowFuncKind, }; use crate::ir::scalar::{Predicate, ProjectItem, ScalarExpr, SortKey}; use crate::ir::schema::aggregate_schema::aggregate_output_schema; @@ -500,6 +501,93 @@ impl NonASAPOp { } impl NonASAPOp { + /// Whether two structurally equal copies of this node may be one shared + /// node: every execution over the same input rows returns the same + /// output rows, so one copy over their one input can be computed (or + /// stored) once and read by both readers. + /// + /// Ties are the one place the input order decides: a `Sort` keeps tied + /// rows in input order (a stable sort), and an exact top-k ranks with that + /// sort. That tie order counts as fixed. An operator that picks rows by + /// input order alone does not return the same rows: + /// - a `Limit` or `OFFSET` whose child is not a `Sort` with the same + /// partitions (`LIMIT 10` may return any 10 rows); + /// - a `Dedup` on some but not all columns (which row of each key + /// survives is unspecified); + /// - a window function that reads row positions (`ROW_NUMBER`, `LAG`, + /// `FIRST_VALUE`, a `ROWS` frame, …) without an `ORDER BY`. + /// + /// Every other operator computes its output from the input rows' values + /// only. `limitk` and `limit_ratio` pick the same series on every + /// evaluation by definition ([`SampleKind`]). This assumes every scalar + /// function the IR accepts is deterministic (no `random()`; unregistered + /// SQL functions are rejected) and that a deployment's + /// `AggIntent::Extension` returns the same rows for the same input. + pub fn is_shareable_when_equal(&self) -> bool { + use NonASAPOp::*; + match self { + Limit { + n, + offset, + partition_by, + child, + } => { + (n.is_none() && *offset == 0) + || matches!(child.non_asap(), Some(Sort { partition_by: sorted_by, .. }) + if sorted_by == partition_by) + } + // Empty `cols` means every column. + Dedup { cols, child } => { + cols.is_empty() + || cols.iter().collect::>().len() + == child.schema.fields.len() + } + SQLWindowFunc { + func, + order_by, + frame, + .. + } => { + let reads_row_positions = match func { + WindowFuncKind::RowNumber + | WindowFuncKind::Lag + | WindowFuncKind::Lead + | WindowFuncKind::LagInFrame + | WindowFuncKind::LeadInFrame + | WindowFuncKind::FirstValue + | WindowFuncKind::LastValue + | WindowFuncKind::NthValue(_) => true, + WindowFuncKind::Rank | WindowFuncKind::DenseRank => false, + WindowFuncKind::Sum + | WindowFuncKind::Avg + | WindowFuncKind::Count + | WindowFuncKind::Min + | WindowFuncKind::Max => frame + .as_ref() + .is_some_and(|frame| frame.units == WindowFrameUnits::Rows), + }; + !reads_row_positions || !order_by.is_empty() + } + Scan { .. } + | Values { .. } + | Filter { .. } + | Project { .. } + | Aggregate { .. } + | Join { .. } + | SetOp { .. } + | Concat { .. } + | Sort { .. } + | BinaryOp { .. } + | TimeRange { .. } + | TimeShift { .. } + | PromqlVectorFromScalar(_) + | PromqlRelabel { .. } + | PromqlInfoEnrich { .. } + | PromqlSeriesSample { .. } + | PromqlSubquery { .. } => true, + } + } + /// Output schema derived from this operator's parameters and its /// children's (already derived) schemas. pub fn output_schema(&self) -> Result { diff --git a/crates/types/src/ir/schema/mod.rs b/crates/types/src/ir/schema/mod.rs index ce15c3f92..1a34d8ba5 100644 --- a/crates/types/src/ir/schema/mod.rs +++ b/crates/types/src/ir/schema/mod.rs @@ -347,9 +347,7 @@ impl Schema { .position(|c| c.name == name && c.table.as_deref() == Some(table)) } - /// Whether this schema has *any* provable unique key — the signal a - /// reuse-aware planning pass needs to decide whether a producer's output - /// can be safely shared across consumers. + /// Whether this schema has *any* provable unique key. pub fn has_unique_key(&self) -> bool { !self.unique_keys.is_empty() } diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 37134791d..fd5acbe63 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -286,14 +286,18 @@ lists `LogicalMemo::from_roots` with Pass 1. `LogicalMemo::from_roots` runs `share_common_sub_dags` (`crates/types/src/ir/cse.rs`) over the frontends' roots and numbers the shared nodes as groups. Two structurally equal nodes become one group only when sharing them is legal, that is, when every reader -of the shared node reads the rows it would read alone: +of the shared node reads the rows it would read alone. Equal nodes are +interned children first, so they always read one input, and a node is legal +to share when every execution over the same input rows returns the same +output rows (`NonASAPOp::is_shareable_when_equal`). A unique +key is not needed. | Equal nodes | Share one group | |---|---| | ASAP nodes (summary states and their evaluations) | Always (equal operator, schema and guarantee) | -| `Scan`s | Always: they read the same source rows | -| Windows directly over a scan, global scalar reductions, exact top-k keeping the series identity | Always: their rows keep a stable identity | -| Any other non-ASAP node | Only when its output schema has a provable unique key (`Schema::has_unique_key`) | +| Non-ASAP nodes that compute their output from the input rows' values: scans, windows, filters, projections, aggregates (exact top-k included), joins, binary operators, `limitk` | Always | +| A `Sort` keeps tied rows in input order (a stable sort); a `Limit` over a `Sort` and an exact top-k rank with that order | Always: the tie order counts as fixed | +| A `Limit` or offset not over a `Sort`, a `Dedup` on some but not all columns, a window function that reads row positions (`ROW_NUMBER`, `LAG`, `FIRST_VALUE`, a `ROWS` frame) without an `ORDER BY` | Never: they pick rows by input order alone, so two copies may return different rows | Equal nodes that may not share stay separate groups, also within one query. They may still read one shared child group. @@ -356,11 +360,10 @@ avg by (job) (quantile_over_time(0.5, data[5m]) / quantile_over_time(0.5, data[5 G1 is one group for its two readers, because a window directly over a scan may be shared. The two `quantile_over_time(0.5, data[5m])` are one group, G2: -the PromQL frontend adds a series-identity column (`$promql_series_identity`) -to every series, so the per-series quantile's output has a unique key -(series identity and timestamp), and equal non-ASAP nodes with a unique key -share one group. A non-ASAP node without a unique key, such as a -`without(..)` grouping, would stay one group per reader. +both compute the same rows from the same input, so one copy serves both +operands. A unique key is not needed (the PromQL planning path happens to +give this output one, series identity and timestamp); an unordered `LIMIT` +would stay one group per reader. **Step 2, Stage 1 Pass 1.** `try_populate_pass1` gives G2 and G4 seven candidates each. G0, G1, G3 and G5 get one each, their own node. G6 gets diff --git a/docs/develop_docs/pre-asap-ir.md b/docs/develop_docs/pre-asap-ir.md index 82ed4438b..d0360cf43 100644 --- a/docs/develop_docs/pre-asap-ir.md +++ b/docs/develop_docs/pre-asap-ir.md @@ -87,9 +87,10 @@ state and only appear below an ASAP operator; a scalar expression that reads suc typing error (`ScalarExpr::scalar_type`), because state must be read out before a value can use it. Column references are positional `ColumnId`s (indexes into the input schema), never names. -`Schema::has_unique_key()` is the legality gate CSE uses: a non-ASAP producer is only shared -across consumers when its row identity is provable. A `Scan` is shared without one: equal scans -read the same source rows, a SQL table's keyless rows included. +CSE shares a non-ASAP producer across consumers when every execution over the same input rows +returns the same output rows (`NonASAPOp::is_shareable_when_equal`); a unique key +is not needed. An unordered `Limit`, a `Dedup` on some but not all columns and a window function +that reads row positions without an `ORDER BY` are never shared. ## Scalar expressions @@ -154,7 +155,7 @@ The result is `Rc`. `lower_promql_workload`, `lower_sql` / `lower_ Workload search then runs structural CSE (`asap_types::ir::cse::share_common_sub_dags`) once across every root: bottom-up hash-consing where the structural hash is only a filter and the typed `PartialEq` decides sharing, following scalar references like any other input, and -gated by `Schema::has_unique_key()` for non-ASAP producers other than scans. +gated by `NonASAPOp::is_shareable_when_equal` for non-ASAP producers. ## Fields and column references