From 086054423e32ba5b7830491be8d9025a66825450 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 01:18:10 +0000 Subject: [PATCH 01/11] fix(optimizer): share identical global scalar aggregates --- .../tests/identical_exact_queries.rs | 64 +++++++++++++++++ .../tests/sharing_issue_fixtures/922.json | 69 +++++++++++++++++++ crates/types/src/ir/cse.rs | 41 +++++++++-- 3 files changed, 169 insertions(+), 5 deletions(-) create mode 100644 crates/integration-tests/tests/identical_exact_queries.rs create mode 100644 crates/integration-tests/tests/sharing_issue_fixtures/922.json diff --git a/crates/integration-tests/tests/identical_exact_queries.rs b/crates/integration-tests/tests/identical_exact_queries.rs new file mode 100644 index 000000000..6a1190f8e --- /dev/null +++ b/crates/integration-tests/tests/identical_exact_queries.rs @@ -0,0 +1,64 @@ +//! Original issue workloads against the production candidate-connection planner. +use asap_devtools::planner_fuzz::{read_seed_file, FuzzSettings, FuzzWorkload, Planner}; +use asap_devtools::planning_stages::run_search_over; +use asap_plan_selection::{PlanSelectionSettings, PlanningModels, Stage1ToStage3Search}; +use asap_planner::PlannerExecutorCompileCheck; +use serde_json::json; +use std::path::Path; + +fn workload() -> FuzzWorkload { + let mut w = read_seed_file( + &Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/sharing_issue_fixtures/922.json"), + ) + .unwrap(); + w.tables = vec![json!({"name":"metrics", "time_index":0, "columns":[ + {"name":"ts","type":"timestamp","nullable":false}, + {"name":"city","type":"utf8","nullable":false}, + {"name":"host","type":"utf8","nullable":false}, + {"name":"value","type":"float64","nullable":false} + ]})]; + w +} + +fn cost(w: &FuzzWorkload) -> f64 { + let roots = Planner::new(FuzzSettings::default()).lower(w).unwrap(); + let planning = w.planning_workload(); + let entries: Vec<_> = planning.query_workload.entries().collect(); + let settings = PlanSelectionSettings::default(); + let deployment = w.deployment(); + let models = PlanningModels::builtin() + .with_plan_selection(&settings) + .with_deployment_capabilities(&deployment) + .with_executor_compile_check(&PlannerExecutorCompileCheck); + let run = run_search_over( + roots, + &entries, + &w.data, + models, + Stage1ToStage3Search { + milp_time_limit_ms: Some(30_000), + ..Default::default() + }, + ) + .unwrap(); + let selection = run.stage3_plan_selection.as_ref().unwrap(); + let value = selection.costs[&selection.selected] + .cost_per_second + .unwrap(); + println!("cost={value}; selected={}", selection.selected); + value +} + +/// #922: two identical exact percentile queries pay for their result once. +#[test] +fn identical_exact_percentiles_share_work() { + let related = workload(); + let mut base = related.clone(); + base.queries.truncate(1); + let base_cost = cost(&base); + let related_cost = cost(&related); + assert!( + related_cost <= base_cost * 1.001, + "{related_cost} vs {base_cost}" + ); +} diff --git a/crates/integration-tests/tests/sharing_issue_fixtures/922.json b/crates/integration-tests/tests/sharing_issue_fixtures/922.json new file mode 100644 index 000000000..997650b02 --- /dev/null +++ b/crates/integration-tests/tests/sharing_issue_fixtures/922.json @@ -0,0 +1,69 @@ +{ + "query_workload": { + "language": { + "sql": "datafusion_sql" + }, + "query_batch": null, + "repeating_queries": [ + { + "query": "SELECT percentile_cont(0.99) WITHIN GROUP (ORDER BY value) AS v FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE", + "demand": { + "fixed_interval": 30000 + }, + "requirements": { + "accuracy": { + "max_error": 0.0, + "failure_probability": 0.0, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": null + }, + "predictability": { + "predictable": { + "known_at": null + } + } + }, + { + "query": "SELECT percentile_cont(0.99) WITHIN GROUP (ORDER BY value) AS v FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE", + "demand": { + "fixed_interval": 30000 + }, + "requirements": { + "accuracy": { + "max_error": 0.0, + "failure_probability": 0.0, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": null + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 1000.0 + } + }, + "sources": { + "metrics": { + "series_count": 10000, + "metric_type": null, + "values_non_negative": true, + "sample_interval": 10000, + "stream_norms": null, + "max_distinct_per_evaluation": null, + "distinct_values_per_label": { + "city": 10, + "host": 100 + } + } + } + } +} diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs index c6d48fb81..06b7cebb9 100644 --- a/crates/types/src/ir/cse.rs +++ b/crates/types/src/ir/cse.rs @@ -38,11 +38,13 @@ //! 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 -//! an ungrouped aggregate, a `without(..)` grouping, a `Concat`/`SetOp` that +//! 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. //! 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. @@ -213,7 +215,21 @@ impl InternTable { // (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 { .. })); + || matches!(node.non_asap(), Some(NonASAPOp::Scan { .. })) + || 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 { .. } + ))); let bucket = self.buckets.entry(hash).or_default(); if reusable { if let Some(existing) = bucket @@ -397,7 +413,7 @@ mod tests { } #[test] - fn no_unique_keys_means_no_merge_even_when_structurally_identical() { + fn identical_global_scalar_aggregates_share_their_single_result() { let a = quantile_agg(vec![], Some(2), 0.9); let b = quantile_agg(vec![], Some(2), 0.9); assert_eq!(a, b, "fixture sanity: structurally equal"); @@ -407,9 +423,24 @@ mod tests { ); let (ra, rb) = two_roots(a, b); assert!( - !Rc::ptr_eq(&ra, &rb), - "no unique key ⇒ never hoisted, even for an identical structural match" + Rc::ptr_eq(&ra, &rb), + "a global scalar aggregate has one stable result per evaluation" + ); + } + + /// A global scalar result can be shared only with the same computation. + #[test] + fn distinct_global_quantiles_do_not_merge() { + let (a, b) = two_roots( + quantile_agg(vec![], Some(2), 0.5), + quantile_agg(vec![], Some(2), 0.9), + ); + assert!(!Rc::ptr_eq(&a, &b)); + let (a, b) = two_roots( + quantile_agg(vec![], Some(2), 0.5), + quantile_agg(vec![], Some(3), 0.5), ); + assert!(!Rc::ptr_eq(&a, &b)); } /// Two equal scans of a keyless table (SQL rows) are one node: each From 76af19a6b023b16f2c50dead2359018f6c186a1c Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 01:48:34 +0000 Subject: [PATCH 02/11] fix(search): seed mixed schedules from maintained state without weakening numerical guards --- .../src/search/refinement/planner.rs | 154 ++++++++++++++++++ 1 file changed, 154 insertions(+) diff --git a/crates/plan-selection/src/search/refinement/planner.rs b/crates/plan-selection/src/search/refinement/planner.rs index 34c21a1af..ec6f4c9c1 100644 --- a/crates/plan-selection/src/search/refinement/planner.rs +++ b/crates/plan-selection/src/search/refinement/planner.rs @@ -150,10 +150,20 @@ pub(in crate::search) fn optimize_configured_candidates( direct.push(seed); } } + // A maintained-state seed avoids the large raw-scan latency range without + // weakening the direct model's numerical guards. It only proposes plans; + // the unrestricted graph and evaluator still determine bounds and validity. + let maintained = joint + .as_ref() + .filter(|(_, profile)| profile.schedules.is_some()) + .and_then(|(_, profile)| maintained_state_profile(configured, profile, context.deadline)); let mut direct_refs: Vec<_> = direct .iter() .map(|(graph, profile)| (graph, DirectPriceModel::Fixed(profile))) .collect(); + if let (Some((graph, _)), Some(profile)) = (&joint, &maintained) { + direct_refs.push((graph, DirectPriceModel::Joint(profile))); + } if let Some((graph, profile)) = &joint { direct_refs.push((graph, DirectPriceModel::Joint(profile))); } @@ -238,6 +248,98 @@ pub(in crate::search) fn optimize_configured_candidates( } /// Every supported local price interface participates in one joint solve. +/// Keep only paths whose scans run at ingestion time. Prune unsupported price +/// tuples to a fixed point before compacting option indices. +fn maintained_state_profile( + configured: &CandidateConfigurationInventory, + profile: &crate::search::milp::direct_price_options::JointPriceProfile, + deadline: Option, +) -> Option { + let mut supported: Vec> = profile + .candidates + .iter() + .enumerate() + .map(|(id, options)| { + let scan = configured.inventory.templates[id] + .as_ref() + .is_some_and(|node| { + matches!( + node.non_asap(), + Some(asap_types::ir::NonASAPOp::Scan { .. }) + ) + }); + options + .iter() + .map(|option| !scan || option.profile.ingestion) + .collect() + }) + .collect(); + loop { + let mut changed = false; + for (id, options) in profile.candidates.iter().enumerate() { + if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) { + return None; + } + for (index, option) in options.iter().enumerate() { + if supported[id][index] + && !option.compatible_input_tuples.iter().any(|tuple| { + tuple + .iter() + .all(|binding| supported[binding.producer.0][binding.price_option]) + }) + { + supported[id][index] = false; + changed = true; + } + } + } + if !changed { + break; + } + } + let remap: Vec>> = supported + .iter() + .map(|options| { + let mut next = 0; + options + .iter() + .map(|keep| { + if *keep { + let index = next; + next += 1; + Some(index) + } else { + None + } + }) + .collect() + }) + .collect(); + let mut result = profile.clone(); + for (id, options) in result.candidates.iter_mut().enumerate() { + let mut index = 0; + options.retain_mut(|option| { + let keep = supported[id][index]; + index += 1; + if keep { + option.compatible_input_tuples.retain(|tuple| { + tuple + .iter() + .all(|binding| supported[binding.producer.0][binding.price_option]) + }); + for tuple in &mut option.compatible_input_tuples { + for binding in tuple { + binding.price_option = + remap[binding.producer.0][binding.price_option].unwrap(); + } + } + } + keep + }); + } + Some(result) +} + fn joint_cost_seed( configured: &CandidateConfigurationInventory, context: &CandidatePlanningContext<'_>, @@ -1119,6 +1221,58 @@ mod tests { ) } + /// Restricting a seed removes all query-time scan paths and repairs every + /// supplier option reference without changing the original profile. + #[test] + fn maintained_seed_prunes_and_remaps_supplier_options() { + use crate::search::costing::direct_profile::extract_price_options; + let (configured, roots, entries, data) = fixture(); + let context = CandidatePlanningContext { + deadline: None, + refine_materialization: false, + original_roots: &roots, + entries: &entries, + data: &data, + models: PlanningModels::builtin(), + calibration: CalibrationFit::default(), + }; + let profile = extract_price_options(&configured, &context) + .unwrap() + .unwrap(); + let original_count: usize = profile.candidates.iter().map(Vec::len).sum(); + let restricted = maintained_state_profile(&configured, &profile, None).unwrap(); + assert!(restricted + .candidates + .iter() + .any(|options| !options.is_empty())); + assert!(restricted.candidates.iter().map(Vec::len).sum::() < original_count); + for (id, options) in restricted.candidates.iter().enumerate() { + let scan = configured.inventory.templates[id] + .as_ref() + .is_some_and(|node| { + matches!( + node.non_asap(), + Some(asap_types::ir::NonASAPOp::Scan { .. }) + ) + }); + for option in options { + assert!(!scan || option.profile.ingestion); + assert!(!option.compatible_input_tuples.is_empty()); + for binding in option.compatible_input_tuples.iter().flatten() { + assert!(binding.price_option < restricted.candidates[binding.producer.0].len()); + } + } + } + assert_eq!( + profile.candidates.iter().map(Vec::len).sum::(), + original_count + ); + assert!( + maintained_state_profile(&configured, &profile, Some(std::time::Instant::now())) + .is_none() + ); + } + // An optional direct-price proposal must not abort planning when its // schedule construction budget expires; the ordinary search remains usable. #[test] From a8b4c6b73f28c579be00a62387d3d114225f3749 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 02:07:09 +0000 Subject: [PATCH 03/11] fix(cse): share source windows and identical series rankings --- .../tests/identical_exact_queries.rs | 9 +++++++++ crates/types/src/ir/cse.rs | 13 +++++++++++++ 2 files changed, 22 insertions(+) diff --git a/crates/integration-tests/tests/identical_exact_queries.rs b/crates/integration-tests/tests/identical_exact_queries.rs index 6a1190f8e..d25f04f12 100644 --- a/crates/integration-tests/tests/identical_exact_queries.rs +++ b/crates/integration-tests/tests/identical_exact_queries.rs @@ -62,3 +62,12 @@ fn identical_exact_percentiles_share_work() { "{related_cost} vs {base_cost}" ); } + +/// Different ranks share the source window but still compute their own values. +#[test] +fn different_exact_percentiles_share_the_input_window() { + let mut related = workload(); + related.queries[1].entry.query.0 = related.queries[1].entry.query.0.replace("0.99", "0.5"); + let related_cost = cost(&related); + assert!(related_cost <= 0.0883, "{related_cost}"); +} diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs index 06b7cebb9..a6a91b768 100644 --- a/crates/types/src/ir/cse.rs +++ b/crates/types/src/ir/cse.rs @@ -45,6 +45,8 @@ //! 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. //! 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,6 +218,17 @@ impl InternTable { 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, From ee7f04804da149c455125887aca5b08e019feadd Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 02:07:09 +0000 Subject: [PATCH 04/11] fix(optimizer): reuse single panes and roll up per-series values and counts --- .../src/asap_aware_cse.rs | 115 ++++++++++++++++-- .../src/memo/rules/rollup.rs | 11 +- 2 files changed, 113 insertions(+), 13 deletions(-) diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs index 09e98fc8a..7185ebfc7 100644 --- a/crates/logical-asap-optimization/src/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs @@ -1189,7 +1189,7 @@ pub fn pane_candidates( ); } let pane_count = u32::try_from(window_ms / width).ok()?; - (!readers.is_empty() && pane_count >= 2).then_some(PaneCandidate { + (!readers.is_empty() && pane_count >= 1).then_some(PaneCandidate { build, pane_width_ms: width, pane_count, @@ -2348,6 +2348,40 @@ fn same_rows(a: &Rc, b: &Rc) -> bool { child: other, }), ) => shift == other_shift && same_rows(child, other), + ( + Some(NonASAPOp::Aggregate { + reduction: Reduction::PerEntity, + measures, + filters, + having: None, + child, + .. + }), + Some(NonASAPOp::Aggregate { + reduction: Reduction::PerEntity, + measures: other_measures, + filters: other_filters, + having: None, + child: other, + .. + }), + ) => { + // These reductions preserve series identity and read the implicit + // sample value. Extra projected labels do not change their values. + // Explicit column indices and filtered measures need remapping and + // are deliberately excluded. + measures == other_measures + && matches!( + measures.as_slice(), + [AggIntent::Avg { col: None } + | AggIntent::Sum { col: None } + | AggIntent::Min { col: None } + | AggIntent::Max { col: None }] + ) + && filters.iter().all(Option::is_none) + && other_filters.iter().all(Option::is_none) + && same_rows(child, other) + } _ => false, } } @@ -2355,8 +2389,8 @@ fn same_rows(a: &Rc, b: &Rc) -> bool { /// The finalized value of `coarse` (`coarse_value`) as an exact /// accumulator over `finer` (the finer build's finalized values), grouped by /// the coarse keys, finalized, with `coarse_value`'s column names; `None` -/// when it cannot be built, the finer values are not Float64 (the only -/// update type exact accumulators execute), or it differs from +/// when it cannot be built, a non-count finer value is not Float64 (the +/// supported accumulator update type), or it differs from /// `coarse_value` in type. fn rolled_up( coarse: &OperatorNode, @@ -2385,8 +2419,33 @@ fn rolled_up( }; // A grouped accumulator outputs its keys, then its value. let value = fine_by.len(); - // Exact accumulators update on Float64 values only: an integer count is - // not summed again. + // Merge count states directly: converting finalized Int64 counts through + // a Float64 sum would lose integer precision above 2^53. + if matches!(family, FieldDataType::ExactAggregate(ExactKind::Count, _)) { + let merged = regrouped(coarse, fine)?; + let finalized = + OperatorNode::new_shared(Operator::ASAP(ASAPOp::FinalizeExactAccumulator { + child: merged, + })) + .ok()?; + if finalized.schema.fields.len() != coarse_value.schema.fields.len() + || finalized + .schema + .fields + .iter() + .zip(&coarse_value.schema.fields) + .any(|(a, b)| a.dtype != b.dtype || a.nullable != b.nullable) + || finalized.schema.unique_keys != coarse_value.schema.unique_keys + { + return None; + } + // Finalization preserves the count's type; only the reader's SQL alias + // differs from the canonical state readout name. + return Some(Rc::new(OperatorNode::with_schema( + finalized.operator.clone(), + coarse_value.schema.clone(), + ))); + } if finer.schema.fields.get(value).map(|f| &f.dtype) != Some(&FieldDataType::Plain( asap_types::ir::schema::DataType::Float64, @@ -3415,8 +3474,8 @@ mod tests { } /// A 5-min window evaluated every 2 min, 90 s or 1 min is built in panes - /// of gcd(window, interval); a query evaluated once per window, or not - /// at a fixed interval, has no panes. + /// of gcd(window, interval). A once-per-window reader has one pane, + /// which can reuse the newest pane of a longer chain. #[test] fn pane_width_is_the_gcd_of_the_window_and_the_interval() { let roots = chose( @@ -3426,7 +3485,7 @@ mod tests { assert_eq!(panes(&roots, &[Some(120_000)]), [(60_000, 5)]); assert_eq!(panes(&roots, &[Some(90_000)]), [(30_000, 10)]); assert_eq!(panes(&roots, &[Some(60_000)]), [(60_000, 5)]); - assert!(panes(&roots, &[Some(300_000)]).is_empty()); + assert_eq!(panes(&roots, &[Some(300_000)]), [(300_000, 1)]); assert!(panes(&roots, &[None]).is_empty()); } @@ -4070,11 +4129,43 @@ mod tests { } } - /// Minima roll up as minima, maxima as maxima. Integer counts do not - /// roll up: exact accumulators execute only Float64 updates. + /// Identical per-series averages retain their values under label projection; + /// different windows, functions, and source predicates must stay separate. + #[test] + fn rollup_crosses_only_identical_per_series_reductions() { + let input = |query: &str| { + let root = lower_promql(query, asap_types::ir::properties::AccuracyTarget::exact()); + OperatorNode::reachable(&root) + .into_iter() + .find(|node| { + matches!( + node.non_asap(), + Some(NonASAPOp::Aggregate { + reduction: Reduction::PerEntity, + .. + }) + ) + }) + .unwrap() + }; + let fine = input("sum by (job, instance) (avg_over_time(m[1m]))"); + assert!(same_rows( + &input("sum by (job) (avg_over_time(m[1m]))"), + &fine + )); + for coarse in [ + "sum by (job) (avg_over_time(m[2m]))", + "sum by (job) (max_over_time(m[1m]))", + "sum by (job) (avg_over_time(m{zone=\"east\"}[1m]))", + ] { + assert!(!same_rows(&input(coarse), &fine), "{coarse}"); + } + } + + /// Minima and maxima roll up as extrema; counts merge their integer states. #[test] - fn extrema_roll_up_and_integer_counts_do_not() { - assert!(rollup_candidates(&exact(&[ + fn extrema_and_integer_counts_roll_up() { + assert!(!rollup_candidates(&exact(&[ "count by (job) (m)", "count by (job, instance) (m)" ])) diff --git a/crates/logical-asap-optimization/src/memo/rules/rollup.rs b/crates/logical-asap-optimization/src/memo/rules/rollup.rs index 9e11bbf81..814ffcc99 100644 --- a/crates/logical-asap-optimization/src/memo/rules/rollup.rs +++ b/crates/logical-asap-optimization/src/memo/rules/rollup.rs @@ -57,6 +57,15 @@ pub(in crate::memo) fn add_rollup_candidates( { continue; } + let state_rollup = OperatorNode::reachable(&rule.rollup).iter().any(|node| { + matches!( + node.asap(), + Some(ASAPOp::SummaryMerge { + group_by: Some(_), + .. + }) + ) + }); let is_sketch = matches!( rule.finer.asap(), Some(ASAPOp::SummaryAgg { @@ -82,7 +91,7 @@ pub(in crate::memo) fn add_rollup_candidates( let inputs = combined_inputs(coarse, fine); let coarse_node = restore_inputs(coarse_node, memo, &inputs); let fine_node = restore_inputs(fine_node, memo, &inputs); - if is_sketch { + if is_sketch || state_rollup { // Concrete resized states become ordinary state groups after all // producer-level configurations have been expanded. added += usize::from(append_reader( From 992cad6142ed93143d6ec90937969533ec031c5f Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 02:07:10 +0000 Subject: [PATCH 05/11] fix(search): validate structural maintained-state seeds when direct pricing is unsupported --- .../src/search/refinement/planner.rs | 50 ++++++++++++++++- .../src/search/refinement/search.rs | 56 +++++++++++++++++++ 2 files changed, 104 insertions(+), 2 deletions(-) diff --git a/crates/plan-selection/src/search/refinement/planner.rs b/crates/plan-selection/src/search/refinement/planner.rs index ec6f4c9c1..e15a1f5a0 100644 --- a/crates/plan-selection/src/search/refinement/planner.rs +++ b/crates/plan-selection/src/search/refinement/planner.rs @@ -157,6 +157,43 @@ pub(in crate::search) fn optimize_configured_candidates( .as_ref() .filter(|(_, profile)| profile.schedules.is_some()) .and_then(|(_, profile)| maintained_state_profile(configured, profile, context.deadline)); + // If even this smaller profile has unrepresentable monetary coefficients, + // its supported paths can still provide a structural feasibility seed. + let maintained_graph = maintained + .as_ref() + .map(|profile| { + use crate::search::candidates::graph::{PhysicalCandidateId, ReaderConstraint}; + let mut graph = configured.inventory.graph.clone(); + for (id, candidate) in graph.candidates.iter_mut().enumerate() { + candidate.local_cost = weights[id]; + if profile.candidates[id].is_empty() { + candidate.producer_requirements.clear(); + graph + .reader_constraints + .push(ReaderConstraint::AllowedOutputs { + candidate: PhysicalCandidateId(id), + outputs: Vec::new(), + }); + } else { + for input in &mut candidate.producer_requirements { + input + .compatible_producers + .retain(|producer| !profile.candidates[producer.0].is_empty()); + } + } + } + for root in &mut graph.query_roots { + root.compatible_candidates + .retain(|id| !profile.candidates[id.0].is_empty()); + } + graph + }) + .filter(|graph| { + graph + .query_roots + .iter() + .all(|root| !root.compatible_candidates.is_empty()) + }); let mut direct_refs: Vec<_> = direct .iter() .map(|(graph, profile)| (graph, DirectPriceModel::Fixed(profile))) @@ -164,6 +201,9 @@ pub(in crate::search) fn optimize_configured_candidates( if let (Some((graph, _)), Some(profile)) = (&joint, &maintained) { direct_refs.push((graph, DirectPriceModel::Joint(profile))); } + if let Some(graph) = &maintained_graph { + direct_refs.push((graph, DirectPriceModel::Structural)); + } if let Some((graph, profile)) = &joint { direct_refs.push((graph, DirectPriceModel::Joint(profile))); } @@ -248,7 +288,9 @@ pub(in crate::search) fn optimize_configured_candidates( } /// Every supported local price interface participates in one joint solve. -/// Keep only paths whose scans run at ingestion time. Prune unsupported price +/// Keep paths whose scans run at ingestion time and whose query readouts are +/// not materialized. This avoids tiny held-memory charges on scalar readouts +/// spanning a much smaller scale than ingestion costs. Prune unsupported price /// tuples to a fixed point before compacting option indices. fn maintained_state_profile( configured: &CandidateConfigurationInventory, @@ -270,7 +312,10 @@ fn maintained_state_profile( }); options .iter() - .map(|option| !scan || option.profile.ingestion) + .map(|option| { + (!scan || option.profile.ingestion) + && (option.profile.ingestion || !option.profile.materialized) + }) .collect() }) .collect(); @@ -1257,6 +1302,7 @@ mod tests { }); for option in options { assert!(!scan || option.profile.ingestion); + assert!(option.profile.ingestion || !option.profile.materialized); assert!(!option.compatible_input_tuples.is_empty()); for binding in option.compatible_input_tuples.iter().flatten() { assert!(binding.price_option < restricted.candidates[binding.producer.0].len()); diff --git a/crates/plan-selection/src/search/refinement/search.rs b/crates/plan-selection/src/search/refinement/search.rs index fa865e5a1..0781e16e0 100644 --- a/crates/plan-selection/src/search/refinement/search.rs +++ b/crates/plan-selection/src/search/refinement/search.rs @@ -11,6 +11,8 @@ use crate::search::milp::direct_price_options::{ pub(in crate::search) enum DirectPriceModel<'a> { Fixed(&'a DirectCostProfile), Joint(&'a JointPriceProfile), + /// Structural ordering only; every price comes from the evaluator. + Structural, } use std::time::{Duration, Instant}; @@ -412,6 +414,21 @@ fn search_with_solver( .is_ok() }) .cloned() + .map(|mut observation| { + if matches!( + restrictions.direct[direct_index].1, + DirectPriceModel::Structural + ) { + observation.selected.modeled_cost = observation + .selected + .selected_candidates + .iter() + .map(|id| restricted.candidates[id.0].local_cost) + .sum(); + observation.exact_cost = None; + } + observation + }) .collect(); ( restricted, @@ -467,6 +484,12 @@ fn search_with_solver( } let proposed = if use_direct { match restrictions.direct[direct_index].1 { + DirectPriceModel::Structural => propose( + proposal_graph, + &proposal_observations, + Some(solve_budget.as_secs_f64()), + &mut solve_metrics, + ), DirectPriceModel::Fixed(profile) => solve_direct_cost_proposal_with_metrics( proposal_graph, &proposal_observations, @@ -647,6 +670,15 @@ fn search_with_solver( .map_or(actual, |previous| previous.min(actual)), ); } + if use_direct + && actual_cost.is_some() + && matches!( + restrictions.direct[direct_index].1, + DirectPriceModel::Structural + ) + { + direct_finished = true; + } observations.push(ObservedAssignment { selected: proposal.selected, exact_cost: actual_cost, @@ -834,6 +866,30 @@ mod tests { assert_eq!(result.solve_metrics.direct_cost_validated_plans, 1); } + /// A structural seed's artificial objective never becomes a monetary + /// bound or a claim that direct pricing matched the evaluator. + #[test] + fn structural_restricted_seed_uses_only_evaluated_prices() { + let graph = choices(&[0.0, 0.0]); + let mut restricted = graph.clone(); + restricted.query_roots[0].compatible_candidates = vec![PhysicalCandidateId(1)]; + restricted.candidates[1].local_cost = 100.0; + let result = search_candidates_with_direct_feedback( + &graph, + limits(1), + &[1.0, 1.0], + &[], + &[], + &[(&restricted, DirectPriceModel::Structural)], + |_| Ok(Some(2.0).into()), + ) + .unwrap(); + assert_eq!(result.incumbent.unwrap().actual_cost, 2.0); + assert_eq!(result.lower_bound, 0.0); + assert_eq!(result.status, CandidateSearchStatus::RefinementLimit); + assert_eq!(result.solve_metrics.direct_cost_validated_plans, 0); + } + // Separate scan-price classes share one budget and retain the best validated DAG. #[test] fn direct_price_classes_preserve_the_best_incumbent() { From d33e35267c606731b9dc07189efd3e63ecb9d6ad Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 02:07:29 +0000 Subject: [PATCH 06/11] fix(optimizer): realize SQL count-ranked heaps directly over raw rows --- .../tests/per_computation_candidates_sql.rs | 23 ++++++++++--------- .../src/per_computation_candidates.rs | 17 ++++++++++---- 2 files changed, 25 insertions(+), 15 deletions(-) diff --git a/crates/integration-tests/tests/per_computation_candidates_sql.rs b/crates/integration-tests/tests/per_computation_candidates_sql.rs index 2affe25c0..51fdf357f 100644 --- a/crates/integration-tests/tests/per_computation_candidates_sql.rs +++ b/crates/integration-tests/tests/per_computation_candidates_sql.rs @@ -222,11 +222,13 @@ async fn heap_updates_over(fields: Vec, sql: &str) -> Vec<(String, String } other => panic!("one item column: {other:?}"), }, - other => panic!("a named column: {other:?}"), + SummaryInputExpr::Constant(value) => value.to_string(), + other => panic!("a named column or constant: {other:?}"), }; let mut updates = Vec::new(); for (t, target) in inventory.targets.iter().enumerate() { - for (c, candidate) in target.choices.candidates.iter().enumerate() { + for c in 0..target.choice_count() { + let (candidate, _) = target.computation_for(c).unwrap(); if !matches!( candidate, ComputationCandidate::Summary( @@ -261,9 +263,9 @@ const COUNT_RANKED_TOPK: &str = "SELECT KEY, count(*) FROM flows GROUP BY KEY ORDER BY count(*) DESC LIMIT 10"; /// An Int64 key is not mistaken for the ranked value: both heap sketches -/// compose, keyed by the key and weighted by the count. +/// compose over raw rows, keyed by the key with unit weight. #[tokio::test] -async fn topk_heap_over_an_int64_key_is_weighted_by_the_count() { +async fn topk_heap_over_an_int64_key_counts_raw_rows() { let updates = heap_updates( "port", DataType::Int64, @@ -274,15 +276,15 @@ async fn topk_heap_over_an_int64_key_is_weighted_by_the_count() { assert!( updates .iter() - .all(|(item, weight)| item == "port" && weight.starts_with("count(")), + .all(|(item, weight)| item == "port" && weight == "1"), "{updates:?}" ); } /// A key named `value` is not the ranked value: the heap sketches are -/// weighted by the count, not by the key. +/// fed raw rows with unit weight. #[tokio::test] -async fn topk_heap_over_a_key_named_value_is_weighted_by_the_count() { +async fn topk_heap_over_a_key_named_value_counts_raw_rows() { let updates = heap_updates( "value", DataType::Utf8, @@ -293,7 +295,7 @@ async fn topk_heap_over_a_key_named_value_is_weighted_by_the_count() { assert!( updates .iter() - .all(|(item, weight)| item == "value" && weight.starts_with("count(")), + .all(|(item, weight)| item == "value" && weight == "1"), "{updates:?}" ); } @@ -301,8 +303,7 @@ async fn topk_heap_over_a_key_named_value_is_weighted_by_the_count() { /// SQL's per-group top-k idiom (`ROW_NUMBER() OVER (PARTITION BY city /// ORDER BY COUNT(*) DESC) … WHERE rn <= 5`) is a heavy-hitter top-k, as /// PromQL `topk by (city)` is: both heap sketches compose, keyed by the -/// host and weighted by the count (the aggregate's one value column, so -/// read as its sample value). +/// host and fed raw rows with unit weight. #[tokio::test] async fn rownumber_per_group_topk_offers_heap_sketches() { let updates = heap_updates_over( @@ -319,7 +320,7 @@ async fn rownumber_per_group_topk_offers_heap_sketches() { assert!( updates .iter() - .all(|(item, weight)| item == "host" && weight == ""), + .all(|(item, weight)| item == "host" && weight == "1"), "{updates:?}" ); } diff --git a/crates/logical-asap-optimization/src/per_computation_candidates.rs b/crates/logical-asap-optimization/src/per_computation_candidates.rs index 59723336a..64eea7a7b 100644 --- a/crates/logical-asap-optimization/src/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/src/per_computation_candidates.rs @@ -173,6 +173,9 @@ pub enum PruneReason { RequiresExact, /// `count` counts rows, so a possibly-null `avg` input changes the result. NullableAvgInput, + /// Frequency sketches do not implement SQL row-count updates. Keep the + /// exact counter; a fused raw-row TopK has its own keyed update contract. + NoRelationalCountSketchRealization, /// The deployment's executor cannot build the summary's family or /// answer the measure's readout from it /// ([`DeploymentCapabilities::supports`]). @@ -730,6 +733,13 @@ fn generate_with_local_candidate_cache( matches!(choice, ComputationCandidate::ExactAccumulator(_)) }); } + if matches!(intent, AggIntent::Count { .. }) + && node.children()[0].result_kind == OperatorResultKind::Relation + { + choices.prune(PruneReason::NoRelationalCountSketchRealization, |choice| { + matches!(choice, ComputationCandidate::Summary(_)) + }); + } targets.push(ComputationTarget { aggregate: node.clone(), measure, @@ -1088,13 +1098,12 @@ fn whole_expression_input( partitions.group_keys().map_or(&[][..], |keys| keys.keys()), )); } else if realized.child.schema.closed - && (matches!(input.item, Some(SummaryInputExpr::EntityIdentity(_))) - || !reads_only_time_series(&realized.child)) + && matches!(input.item, Some(SummaryInputExpr::EntityIdentity(_))) { // An encoded label set needs an open PromQL schema. PromQL rows that // carry the series identity are closed, but their inner group labels - // are item columns. (A SQL top-k keeps its heaps over its per-key - // result.) + // are item columns. SQL group keys are explicit columns too, so their + // keyed heap can consume raw rows without a per-key aggregate first. return None; } Some((realized.child, input, partitions)) From 6a2cf55aa2685f4f4e540a34971bf261ad07da5a Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 02:07:29 +0000 Subject: [PATCH 07/11] fix(optimizer): reuse a filtered grouped quantile with SQL empty-input semantics --- .../src/memo/rules.rs | 1 + .../src/memo/rules/filtered_group.rs | 139 ++++++++++++++++++ .../src/memo/rules/reuse.rs | 7 +- .../src/memo/state_groups.rs | 1 + .../src/per_computation_candidates.rs | 75 ++++++++++ crates/types/src/ir/operator/asap.rs | 10 +- 6 files changed, 228 insertions(+), 5 deletions(-) create mode 100644 crates/logical-asap-optimization/src/memo/rules/filtered_group.rs diff --git a/crates/logical-asap-optimization/src/memo/rules.rs b/crates/logical-asap-optimization/src/memo/rules.rs index da3d586b5..69dd57f49 100644 --- a/crates/logical-asap-optimization/src/memo/rules.rs +++ b/crates/logical-asap-optimization/src/memo/rules.rs @@ -1,5 +1,6 @@ //! Bounded logical memo rule expansion. Rules expose computations and typed //! dependencies; physical configuration and cost selection live downstream. +pub(super) mod filtered_group; pub(super) mod reuse; pub(super) mod rollup; pub(super) mod scan_labels; diff --git a/crates/logical-asap-optimization/src/memo/rules/filtered_group.rs b/crates/logical-asap-optimization/src/memo/rules/filtered_group.rs new file mode 100644 index 000000000..124acbb0a --- /dev/null +++ b/crates/logical-asap-optimization/src/memo/rules/filtered_group.rs @@ -0,0 +1,139 @@ +//! Read one existing grouped quantile instead of rebuilding its filtered input. +use super::reuse::append_reader; +use crate::memo::candidates::{CandidateMemo, CandidateOrigin}; +use crate::memo::check_generation_deadline; +use crate::memo::state_groups::StateGroupExpansionError; +use asap_types::ir::operator::{AggIntent, Reduction}; +use asap_types::ir::scalar::CompareOpKind; +use asap_types::ir::{NonASAPOp, Operator, OperatorNode, Predicate, ScalarExpr}; +use std::rc::Rc; +use std::time::Instant; + +// Removing exactly one key equality must leave the identical source and window. +fn extra_key_filter(a: &Rc, b: &Rc, key: usize) -> Option { + match (a.non_asap()?, b.non_asap()?) { + ( + NonASAPOp::TimeRange { range, kind, child }, + NonASAPOp::TimeRange { + range: r, + kind: k, + child: other, + }, + ) if range == r && kind == k => extra_key_filter(child, other, key), + ( + NonASAPOp::Scan { + source, + schema, + predicates, + }, + NonASAPOp::Scan { + source: other, + schema: s, + predicates: base, + }, + ) if source == other && schema == s && predicates.len() == base.len() + 1 => { + for (index, pred) in predicates.iter().enumerate() { + if predicates + .iter() + .enumerate() + .filter(|(i, _)| *i != index) + .map(|(_, p)| p) + .eq(base.iter()) + && matches!(&pred.0, ScalarExpr::Compare { left, op: CompareOpKind::Eq, right, .. } + if matches!(&**left, ScalarExpr::Column(i) if *i == key) && matches!(&**right, ScalarExpr::Literal(_))) + { + return Some(Predicate(pred.0.map_columns(&mut |_| 0))); + } + } + None + } + _ => None, + } +} + +pub(in crate::memo) fn add_filtered_group_candidates( + memo: &mut CandidateMemo, + deadline: Option, +) -> Result { + let groups = memo.groups.clone(); + let mut added = 0; + for (owner, coarse) in groups.iter().enumerate() { + check_generation_deadline(deadline)?; + if coarse.node.result_kind != asap_types::ir::OperatorResultKind::Relation { + continue; + } + let Some(NonASAPOp::Aggregate { + reduction: Reduction::Reduce(by), + measures, + filters, + having: None, + child, + .. + }) = coarse.node.non_asap() + else { + continue; + }; + if by.is_without() + || !by.is_empty() + || !filters.iter().all(Option::is_none) + || !matches!(measures.as_slice(), [AggIntent::Quantile { .. }]) + { + continue; + } + for fine in &groups { + check_generation_deadline(deadline)?; + let Some(NonASAPOp::Aggregate { + reduction: Reduction::Reduce(keys), + measures: other, + filters: other_filters, + having: None, + child: source, + .. + }) = fine.node.non_asap() + else { + continue; + }; + if keys.is_without() + || keys.len() != 1 + || measures != other + || !other_filters.iter().all(Option::is_none) + { + continue; + } + let Some(pred) = extra_key_filter(child, source, keys.keys()[0]) else { + continue; + }; + let filtered = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Filter { + pred, + child: Rc::clone(&fine.node), + })) + .map_err(|e| StateGroupExpansionError::ReuseRule(e.to_string()))?; + // SQL's global quantile returns one NULL row on an empty input. + // MIN over the selected (at most one) group preserves that contract. + let reader = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + reduction: Reduction::by(vec![]), + measures: vec![AggIntent::Min { col: Some(1) }], + output_names: vec![coarse.node.schema.fields[0].name.clone()], + filters: vec![], + having: None, + child: filtered, + })) + .map_err(|e| StateGroupExpansionError::ReuseRule(e.to_string()))?; + let Some(original) = memo.candidates[owner] + .iter() + .find(|c| c.origin == CandidateOrigin::Pass1) + .cloned() + else { + continue; + }; + added += usize::from(append_reader( + memo, + owner, + &original, + reader, + CandidateOrigin::Rollup, + )?); + } + } + Ok(added) +} diff --git a/crates/logical-asap-optimization/src/memo/rules/reuse.rs b/crates/logical-asap-optimization/src/memo/rules/reuse.rs index 67ac8e592..196ec77ad 100644 --- a/crates/logical-asap-optimization/src/memo/rules/reuse.rs +++ b/crates/logical-asap-optimization/src/memo/rules/reuse.rs @@ -150,9 +150,10 @@ pub(in crate::memo) fn append_reader( || (original.node.schema.closed && !node.schema.closed) || original.node.result_kind != node.result_kind { - return Err(StateGroupExpansionError::ReuseRule( - "reuse changed the reader output contract".into(), - )); + return Err(StateGroupExpansionError::ReuseRule(format!( + "reuse changed the reader output contract ({origin:?}): expected {:?}, got {:?}", + original.node.schema, node.schema + ))); } let boundaries = memo .groups diff --git a/crates/logical-asap-optimization/src/memo/state_groups.rs b/crates/logical-asap-optimization/src/memo/state_groups.rs index 212aebb57..5f4c5494e 100644 --- a/crates/logical-asap-optimization/src/memo/state_groups.rs +++ b/crates/logical-asap-optimization/src/memo/state_groups.rs @@ -102,6 +102,7 @@ fn expand_with_state_index( super::rules::scan_labels::add_scan_label_candidates(&mut memo, deadline)?; super::rules::summary_inputs::factor_summary_inputs(&mut memo, deadline)?; super::rules::topk::add_topk_candidates(&mut memo, deadline)?; + super::rules::filtered_group::add_filtered_group_candidates(&mut memo, deadline)?; super::rules::rollup::add_rollup_candidates(&mut memo, deadline)?; super::rules::rollup::add_window_rollup_candidates(&mut memo, deadline)?; super::rules::summary_results::factor_summary_results(&mut memo, deadline)?; diff --git a/crates/logical-asap-optimization/src/per_computation_candidates.rs b/crates/logical-asap-optimization/src/per_computation_candidates.rs index 64eea7a7b..a60528aa7 100644 --- a/crates/logical-asap-optimization/src/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/src/per_computation_candidates.rs @@ -1656,6 +1656,78 @@ fn with_guarantees( Ok(rebuilt) } +/// An extremum of at most one selected group is an identity on its value. +/// Unlike a general aggregate over rank estimates, this does not change the +/// population to which the input's error bound refers. +fn selected_group_guarantee(node: &OperatorNode) -> Option { + let NonASAPOp::Aggregate { + reduction: Reduction::Reduce(by), + measures, + filters, + having: None, + child, + .. + } = node.non_asap()? + else { + return None; + }; + if by.is_without() + || !by.is_empty() + || !filters.iter().all(Option::is_none) + || !matches!( + measures.as_slice(), + [AggIntent::Min { col: Some(1) } | AggIntent::Max { col: Some(1) }] + ) + { + return None; + } + let NonASAPOp::Filter { + pred, + child: grouped, + } = child.non_asap()? + else { + return None; + }; + if !matches!(&pred.0, ScalarExpr::Compare { left, op: asap_types::ir::scalar::CompareOpKind::Eq, right, .. } + if matches!(&**left, ScalarExpr::Column(0)) && matches!(&**right, ScalarExpr::Literal(_))) + || !grouped.schema.unique_keys.contains(&vec![0]) + { + return None; + } + let mut source = grouped; + while let Some(NonASAPOp::Project { cols, child, .. }) = source.non_asap() { + if !cols + .iter() + .enumerate() + .all(|(index, item)| matches!(item.expr, ScalarExpr::Column(column) if column == index)) + { + return None; + } + source = child; + } + let Some(ASAPOp::SummaryEstimate { summary_input, .. }) = source.asap() else { + return None; + }; + let Some(ASAPOp::SummaryAgg { + reduction: Reduction::Reduce(keys), + child: raw, + .. + }) = summary_input.asap() + else { + return None; + }; + if keys.is_without() + || keys.len() != 1 + || !raw + .guarantee + .as_ref() + .is_some_and(AccuracyGuarantee::is_exact) + { + return None; + } + grouped.guarantee.clone() +} + /// The theoretical guarantee of a composed node, given its children's, /// composed with the built-in analytical rules ([`propagate`]); /// Stage 1 reads no data statistics or deployment evidence. Exact over exact @@ -1676,6 +1748,9 @@ fn with_guarantees( /// /// [`AccuracyScope::PerValue`]: asap_types::ir::properties::AccuracyScope::PerValue fn composed_guarantee(node: &OperatorNode) -> Option { + if let Some(guarantee) = selected_group_guarantee(node) { + return Some(guarantee); + } if let Some(guarantee) = rolled_up_guarantee(node) { return guarantee; } diff --git a/crates/types/src/ir/operator/asap.rs b/crates/types/src/ir/operator/asap.rs index fe0200eab..8b8683bb8 100644 --- a/crates/types/src/ir/operator/asap.rs +++ b/crates/types/src/ir/operator/asap.rs @@ -981,8 +981,14 @@ fn topk_estimate_schema(state: &OperatorNode) -> Option { /// NULLs gives an empty state). fn quantile_may_be_empty(state: &OperatorNode) -> bool { match state.asap() { - Some(ASAPOp::SummaryMerge { children, .. }) => { - children.iter().any(|child| quantile_may_be_empty(child)) + Some(ASAPOp::SummaryMerge { + children, group_by, .. + }) => { + // A global rollup must also represent the empty input population, + // even when every existing grouped state is nonempty. + (source_kind(state) == OperatorResultKind::Relation + && group_by.as_ref().is_some_and(Vec::is_empty)) + || children.iter().any(|child| quantile_may_be_empty(child)) } Some( ASAPOp::SummaryAgg { From 440dc259e47818caadc1009709367cc0ebbb96d5 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 02:10:54 +0000 Subject: [PATCH 08/11] test(optimizer): cover sharing regressions and SQL execution boundaries --- .../tests/sharing_issue_fixtures/886.json | 39 ++ .../tests/sharing_issue_fixtures/902.json | 73 +++ .../tests/sharing_issue_fixtures/923.json | 81 +++ .../tests/sharing_issue_fixtures/924.json | 81 +++ .../tests/sharing_issue_fixtures/925.json | 81 +++ .../tests/sharing_issue_regressions.rs | 562 ++++++++++++++++++ 6 files changed, 917 insertions(+) create mode 100644 crates/integration-tests/tests/sharing_issue_fixtures/886.json create mode 100644 crates/integration-tests/tests/sharing_issue_fixtures/902.json create mode 100644 crates/integration-tests/tests/sharing_issue_fixtures/923.json create mode 100644 crates/integration-tests/tests/sharing_issue_fixtures/924.json create mode 100644 crates/integration-tests/tests/sharing_issue_fixtures/925.json create mode 100644 crates/integration-tests/tests/sharing_issue_regressions.rs diff --git a/crates/integration-tests/tests/sharing_issue_fixtures/886.json b/crates/integration-tests/tests/sharing_issue_fixtures/886.json new file mode 100644 index 000000000..259fafd9d --- /dev/null +++ b/crates/integration-tests/tests/sharing_issue_fixtures/886.json @@ -0,0 +1,39 @@ +{ + "query_workload": { + "language": { + "sql": "datafusion_sql" + }, + "query_batch": null, + "repeating_queries": [ + { + "query": "SELECT host, COUNT(*) AS c FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE GROUP BY host ORDER BY c DESC LIMIT 5", + "demand": { + "fixed_interval": 10000 + }, + "requirements": { + "accuracy": { + "max_error": 0.01, + "failure_probability": 0.001 + } + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 666.67 + } + }, + "sources": { + "metrics": { + "series_count": 100 + } + } + } +} diff --git a/crates/integration-tests/tests/sharing_issue_fixtures/902.json b/crates/integration-tests/tests/sharing_issue_fixtures/902.json new file mode 100644 index 000000000..e0073ccc8 --- /dev/null +++ b/crates/integration-tests/tests/sharing_issue_fixtures/902.json @@ -0,0 +1,73 @@ +{ + "query_workload": { + "language": { + "sql": "datafusion_sql" + }, + "query_batch": null, + "repeating_queries": [ + { + "query": "SELECT percentile_cont(0.9) WITHIN GROUP (ORDER BY value) FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE", + "demand": { + "fixed_interval": 10000 + }, + "requirements": { + "accuracy": { + "max_error": 0.01, + "failure_probability": 0.001 + } + }, + "predictability": { + "predictable": { + "known_at": null + } + } + }, + { + "query": "SELECT percentile_cont(0.9) WITHIN GROUP (ORDER BY value) FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE AND city = 'NYC'", + "demand": { + "fixed_interval": 10000 + }, + "requirements": { + "accuracy": { + "max_error": 0.01, + "failure_probability": 0.001 + } + }, + "predictability": { + "predictable": { + "known_at": null + } + } + }, + { + "query": "SELECT city, percentile_cont(0.9) WITHIN GROUP (ORDER BY value) FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE GROUP BY city", + "demand": { + "fixed_interval": 10000 + }, + "requirements": { + "accuracy": { + "max_error": 0.01, + "failure_probability": 0.001 + } + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 666.67 + } + }, + "sources": { + "metrics": { + "series_count": 100 + } + } + } +} diff --git a/crates/integration-tests/tests/sharing_issue_fixtures/923.json b/crates/integration-tests/tests/sharing_issue_fixtures/923.json new file mode 100644 index 000000000..89cfe117d --- /dev/null +++ b/crates/integration-tests/tests/sharing_issue_fixtures/923.json @@ -0,0 +1,81 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "sum by (instance, region) (avg_over_time(m_gauge[1m]))", + "demand": { + "fixed_interval": 10000 + }, + "requirements": { + "accuracy": { + "max_error": 0.0, + "failure_probability": 0.0, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": null + }, + "predictability": { + "predictable": { + "known_at": null + } + } + }, + { + "query": "sum by (instance) (avg_over_time(m_gauge[1m]))", + "demand": { + "fixed_interval": 10000 + }, + "requirements": { + "accuracy": { + "max_error": 0.0, + "failure_probability": 0.0, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": null + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 200.0 + } + }, + "sources": { + "m_counter": { + "series_count": 1000, + "metric_type": "counter", + "values_non_negative": true, + "sample_interval": 10000, + "stream_norms": null, + "max_distinct_per_evaluation": null, + "distinct_values_per_label": { + "instance": 50, + "job": 5, + "region": 3 + } + }, + "m_gauge": { + "series_count": 1000, + "metric_type": "gauge", + "values_non_negative": true, + "sample_interval": 10000, + "stream_norms": null, + "max_distinct_per_evaluation": null, + "distinct_values_per_label": { + "instance": 50, + "job": 5, + "region": 3 + } + } + } + } +} diff --git a/crates/integration-tests/tests/sharing_issue_fixtures/924.json b/crates/integration-tests/tests/sharing_issue_fixtures/924.json new file mode 100644 index 000000000..69ed9e5e6 --- /dev/null +++ b/crates/integration-tests/tests/sharing_issue_fixtures/924.json @@ -0,0 +1,81 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "sum (sum_over_time(m_gauge[2m]))", + "demand": { + "fixed_interval": 60000 + }, + "requirements": { + "accuracy": { + "max_error": 0.0, + "failure_probability": 0.0, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": null + }, + "predictability": { + "predictable": { + "known_at": null + } + } + }, + { + "query": "sum (sum_over_time(m_gauge[1m]))", + "demand": { + "fixed_interval": 60000 + }, + "requirements": { + "accuracy": { + "max_error": 0.0, + "failure_probability": 0.0, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": null + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 1333.3333333333333 + } + }, + "sources": { + "m_counter": { + "series_count": 10000, + "metric_type": "counter", + "values_non_negative": true, + "sample_interval": 15000, + "stream_norms": null, + "max_distinct_per_evaluation": null, + "distinct_values_per_label": { + "instance": 50, + "job": 5, + "region": 3 + } + }, + "m_gauge": { + "series_count": 10000, + "metric_type": "gauge", + "values_non_negative": true, + "sample_interval": 15000, + "stream_norms": null, + "max_distinct_per_evaluation": null, + "distinct_values_per_label": { + "instance": 50, + "job": 5, + "region": 3 + } + } + } + } +} diff --git a/crates/integration-tests/tests/sharing_issue_fixtures/925.json b/crates/integration-tests/tests/sharing_issue_fixtures/925.json new file mode 100644 index 000000000..d8dd7b0d6 --- /dev/null +++ b/crates/integration-tests/tests/sharing_issue_fixtures/925.json @@ -0,0 +1,81 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "sum (sum_over_time(m_gauge[2m]))", + "demand": { + "fixed_interval": 60000 + }, + "requirements": { + "accuracy": { + "max_error": 0.0, + "failure_probability": 0.0, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": null + }, + "predictability": { + "predictable": { + "known_at": null + } + } + }, + { + "query": "sum (sum_over_time(m_gauge[2m]))", + "demand": { + "fixed_interval": 120000 + }, + "requirements": { + "accuracy": { + "max_error": 0.0, + "failure_probability": 0.0, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": null + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 1333.3333333333333 + } + }, + "sources": { + "m_counter": { + "series_count": 10000, + "metric_type": "counter", + "values_non_negative": true, + "sample_interval": 15000, + "stream_norms": null, + "max_distinct_per_evaluation": null, + "distinct_values_per_label": { + "instance": 50, + "job": 5, + "region": 3 + } + }, + "m_gauge": { + "series_count": 10000, + "metric_type": "gauge", + "values_non_negative": true, + "sample_interval": 15000, + "stream_norms": null, + "max_distinct_per_evaluation": null, + "distinct_values_per_label": { + "instance": 50, + "job": 5, + "region": 3 + } + } + } + } +} diff --git a/crates/integration-tests/tests/sharing_issue_regressions.rs b/crates/integration-tests/tests/sharing_issue_regressions.rs new file mode 100644 index 000000000..91ef2cfa7 --- /dev/null +++ b/crates/integration-tests/tests/sharing_issue_regressions.rs @@ -0,0 +1,562 @@ +//! Original issue workloads against the production candidate-connection planner. +use asap_devtools::planner_fuzz::{read_seed_file, FuzzSettings, FuzzWorkload, Planner}; +use asap_devtools::planning_stages::run_search_over; +use asap_plan_selection::{PlanSelectionSettings, PlanningModels, Stage1ToStage3Search}; +use asap_planner::PlannerExecutorCompileCheck; +use serde_json::json; +use std::path::Path; + +fn workload(issue: u32) -> FuzzWorkload { + let mut w = read_seed_file( + &Path::new(env!("CARGO_MANIFEST_DIR")) + .join(format!("tests/sharing_issue_fixtures/{issue}.json")), + ) + .unwrap(); + w.tables = vec![json!({"name":"metrics", "time_index":0, "columns":[ + {"name":"ts","type":"timestamp","nullable":false}, + {"name":"city","type":"utf8","nullable":false}, + {"name":"host","type":"utf8","nullable":false}, + {"name":"value","type":"float64","nullable":false} + ]})]; + if issue == 902 || issue == 886 { + for field in w.tables[0]["columns"] + .as_array_mut() + .unwrap() + .iter_mut() + .skip(1) + { + field["nullable"] = json!(true); + } + } + w +} + +fn cost(w: &FuzzWorkload) -> f64 { + let roots = Planner::new(FuzzSettings::default()).lower(w).unwrap(); + let planning = w.planning_workload(); + let entries: Vec<_> = planning.query_workload.entries().collect(); + let settings = PlanSelectionSettings::default(); + let deployment = w.deployment(); + let models = PlanningModels::builtin() + .with_plan_selection(&settings) + .with_deployment_capabilities(&deployment) + .with_executor_compile_check(&PlannerExecutorCompileCheck); + let run = run_search_over( + roots, + &entries, + &w.data, + models, + Stage1ToStage3Search { + milp_time_limit_ms: Some(30_000), + ..Default::default() + }, + ) + .unwrap(); + let selection = run.stage3_plan_selection.as_ref().unwrap(); + let value = selection.costs[&selection.selected] + .cost_per_second + .unwrap(); + println!("cost={value}; selected={}", selection.selected); + value +} + +/// #923: deriving the coarse sum costs only a sum over the 150 fine groups. +#[test] +fn exact_coarse_sum_reuses_fine_sum() { + let related = workload(923); + let mut base = related.clone(); + base.queries.truncate(1); + let base_cost = cost(&base); + let related_cost = cost(&related); + assert!( + related_cost <= base_cost + 0.00004, + "{related_cost} vs {base_cost}" + ); +} + +/// #924: the 1m sum reads the newest pane of the existing 2m chain. +#[test] +fn nested_exact_sum_reuses_newest_panes() { + let related = workload(924); + let mut base = related.clone(); + base.queries.truncate(1); + let base_cost = cost(&base); + let related_cost = cost(&related); + assert!( + related_cost <= base_cost * 1.001, + "{related_cost} vs {base_cost}" + ); +} + +/// #925: the slower copy reads the same maintained sum. +#[test] +fn slower_repeat_reuses_faster_sum() { + let related = workload(925); + let mut base = related.clone(); + base.queries.truncate(1); + let base_cost = cost(&base); + let related_cost = cost(&related); + assert!( + related_cost <= base_cost * 1.001, + "{related_cost} vs {base_cost}" + ); +} + +/// #902: the filtered reader avoids its independent scan and aggregation. +#[test] +fn grouped_filtered_quantile_plans() { + assert!( + cost(&workload(902)) < 0.25, + "filtered query must reuse the grouped result" + ); +} + +/// #886: SQL count-ranked TopK offers a heap over raw rows, absorbing the count. +#[test] +fn sql_count_topk_has_a_fused_raw_row_heap() { + use asap_logical_asap_optimization::per_computation_candidates::generate_per_computation_candidates; + let roots = Planner::new(FuzzSettings::default()) + .lower(&workload(886)) + .unwrap(); + let inventory = + generate_per_computation_candidates(roots.into_iter().enumerate().collect()).unwrap(); + let (index, target) = inventory + .targets + .iter() + .enumerate() + .find(|(_, target)| target.fused.is_some()) + .expect("SQL TopK must offer a fused raw-row heap"); + for choice in target.choices.candidates.len()..target.choice_count() { + let mut selected = vec![0; inventory.targets.len()]; + selected[index] = choice; + let composed = + asap_logical_asap_optimization::per_computation_candidates::compose_logical_candidate( + &inventory, &selected, + ) + .unwrap(); + let roots: Vec<_> = composed.into_iter().map(|(_, root)| root).collect(); + let root = roots[0].as_operator().unwrap(); + for node in asap_types::ir::OperatorNode::reachable(root) { + if let Some(asap_types::ir::ASAPOp::SummaryAgg { + family: asap_types::ir::schema::FieldDataType::Sketch(_, _), + input, + child, + .. + }) = node.asap() + { + assert_eq!( + input.weight, + asap_types::ir::schema::SummaryInputExpr::Constant(1.0) + ); + assert!(!child.contains_asap(), "heap consumes raw rows"); + } + } + let guarded = Planner::new(FuzzSettings::default()).price(&workload(886), &roots, false); + assert!( + guarded.unwrap_err().contains("NoAccuracyGuarantee"), + "frequency intervals must not be claimed as Top-K membership guarantees" + ); + use asap_executor::values::Value; + let rows = (0..7) + .map(|_| { + vec![ + Value::Timestamp(0), + Value::Utf8("NYC".into()), + Value::Utf8("one-host".into()), + Value::Float64(1.0), + ] + }) + .collect(); + let actual = physical_common::execute_raw_rows_at(root, rows, 1000); + assert_eq!(actual.len(), 1, "{actual:?}"); + assert!( + actual[0] + .iter() + .any(|v| matches!(v, Value::Utf8(s) if s.as_ref() == "one-host")), + "{actual:?}" + ); + assert!( + actual[0] + .iter() + .any(|v| matches!(v, Value::Float64(count) if *count == 7.0)), + "{actual:?}" + ); + let compiled = Planner::new(FuzzSettings { + enforce_accuracy_guarantees: false, + ..Default::default() + }) + .price(&workload(886), &roots, false); + assert!(compiled.is_ok(), "fused heap must compile: {compiled:?}"); + } +} + +/// #923: label projection differences below the same per-series average do not +/// prevent an exact sum from rolling up the finer sum's output. +#[test] +fn exact_rollup_catalog_crosses_identical_per_series_average() { + use asap_logical_asap_optimization::{ + memo::{ + candidates::{try_populate_pass1, CandidateOrigin}, + state_configurations::PaneWidthDomain, + state_groups::expand_concrete_summary_state_groups_in_domain, + LogicalMemo, + }, + per_computation_candidates::generate_per_computation_candidates, + }; + let roots: Vec<_> = Planner::new(FuzzSettings::default()) + .lower(&workload(923)) + .unwrap() + .into_iter() + .enumerate() + .collect(); + let initial = LogicalMemo::from_roots(&roots); + let inventory = generate_per_computation_candidates(roots).unwrap(); + let memo = try_populate_pass1(initial, &inventory).unwrap(); + let expanded = + expand_concrete_summary_state_groups_in_domain(memo, PaneWidthDomain::SeedGrids).unwrap(); + let owner = expanded + .memo + .groups + .iter() + .position(|g| g.roots_of.contains(&1)) + .unwrap(); + assert!( + expanded.memo.candidates[owner] + .iter() + .any(|c| c.origin == CandidateOrigin::Rollup), + "coarse root has no rollup candidate" + ); +} + +mod physical_common; + +/// #902: selecting one group preserves values and SQL's one NULL row when +/// the group is absent, including empty inputs and groups of NULL samples. +#[test] +fn filtered_group_preserves_sql_empty_and_null_semantics() { + use asap_executor::values::Value; + use asap_logical_asap_optimization::{ + memo::{ + candidates::{try_populate_pass1, CandidateOrigin}, + state_configurations::PaneWidthDomain, + state_groups::expand_concrete_summary_state_groups_in_domain, + LogicalMemo, + }, + per_computation_candidates::generate_per_computation_candidates, + }; + use asap_types::ir::{operator::AggIntent, NonASAPOp}; + let roots: Vec<_> = Planner::new(FuzzSettings::default()) + .lower(&workload(902)) + .unwrap() + .into_iter() + .enumerate() + .collect(); + let initial = LogicalMemo::from_roots(&roots); + let inventory = generate_per_computation_candidates(roots).unwrap(); + let memo = try_populate_pass1(initial, &inventory).unwrap(); + let expanded = + expand_concrete_summary_state_groups_in_domain(memo, PaneWidthDomain::SeedGrids).unwrap(); + let (owner, candidate) = expanded.memo.candidates.iter().enumerate().flat_map(|(owner, candidates)| candidates.iter().map(move |c| (owner,c))) + .find(|(_, c)| c.origin == CandidateOrigin::Rollup && matches!(c.node.non_asap(), Some(NonASAPOp::Aggregate { measures, child, .. }) + if matches!(measures.as_slice(), [AggIntent::Min { .. }]) && matches!(child.non_asap(), Some(NonASAPOp::Filter { .. })))) + .expect("filtered group read candidate"); + // The same readout keeps a KLL's rank bound: it selects a whole group, + // never predicates on an approximate value or combines populations. + use asap_logical_asap_optimization::per_computation_candidates::{ + compose_logical_candidate, recompose_guarantees, ComputationCandidate, + }; + use asap_types::ir::{schema::SketchAlgorithm, QueryRoot}; + use std::rc::Rc; + let filter = candidate.node.children()[0]; + let grouped = filter.children()[0]; + let inv = + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(Rc::clone(grouped)))]) + .unwrap(); + let choices: Vec<_> = inv + .targets + .iter() + .map(|target| { + target + .choices + .candidates + .iter() + .position(|c| matches!(c, ComputationCandidate::Summary(SketchAlgorithm::Kll))) + .unwrap_or(0) + }) + .collect(); + let approximate = compose_logical_candidate(&inv, &choices) + .unwrap() + .remove(0) + .1; + let approximate = approximate.as_operator().unwrap(); + let filtered = Rc::new( + filter + .with_new_children(|_| Rc::clone(approximate)) + .unwrap(), + ); + let readout = Rc::new( + candidate + .node + .with_new_children(|_| Rc::clone(&filtered)) + .unwrap(), + ); + let recomposed = recompose_guarantees(&[QueryRoot::Operator(readout)]).unwrap(); + assert!(approximate.guarantee.is_some()); + assert_eq!( + recomposed[0].as_operator().unwrap().guarantee, + approximate.guarantee + ); + // A range of keys can select multiple populations; their extrema do not + // inherit one selected quantile's rank bound. + let mut many_groups = (*filtered).clone(); + if let asap_types::ir::Operator::NonASAP(NonASAPOp::Filter { pred, .. }) = + &mut many_groups.operator + { + if let asap_types::ir::ScalarExpr::Compare { op, .. } = &mut pred.0 { + *op = asap_types::ir::scalar::CompareOpKind::Gt; + } + } + let many_groups = Rc::new(many_groups); + let invalid = Rc::new( + candidate + .node + .with_new_children(|_| Rc::clone(&many_groups)) + .unwrap(), + ); + assert!( + recompose_guarantees(&[QueryRoot::Operator(invalid)]).unwrap()[0] + .as_operator() + .unwrap() + .guarantee + .is_none() + ); + let row = |city: &str, value| { + vec![ + Value::Timestamp(0), + Value::Utf8(city.into()), + Value::Utf8("host".into()), + value, + ] + }; + for rows in [ + vec![], + vec![row("LA", Value::Float64(5.0))], + vec![row("NYC", Value::Null)], + vec![ + row("NYC", Value::Float64(10.0)), + row("NYC", Value::Float64(30.0)), + row("LA", Value::Float64(90.0)), + ], + ] { + let before = physical_common::execute_raw_rows_at( + &expanded.memo.groups[owner].node, + rows.clone(), + 1000, + ); + let after = physical_common::execute_raw_rows_at(&candidate.node, rows, 1000); + assert_eq!(format!("{after:?}"), format!("{before:?}")); + assert_eq!(after.len(), 1); + } +} + +/// #923: SQL counts roll up by merging integer states, without float casts. +#[test] +fn sql_count_rollup_executes_with_integer_results() { + use asap_executor::values::Value; + use asap_logical_asap_optimization::{ + memo::{ + candidates::{try_populate_pass1, CandidateOrigin}, + state_configurations::PaneWidthDomain, + state_groups::expand_concrete_summary_state_groups_in_domain, + LogicalMemo, + }, + per_computation_candidates::generate_per_computation_candidates, + }; + use asap_types::ir::ASAPOp; + let mut w = workload(902); + w.queries.truncate(2); + w.queries[0].entry.query.0 = + "SELECT city, host, COUNT(*) AS c FROM metrics GROUP BY city, host".into(); + for query in [ + "SELECT city, COUNT(*) AS c FROM metrics GROUP BY city", + "SELECT COUNT(*) AS c FROM metrics", + ] { + w.queries[1].entry.query.0 = query.into(); + let roots: Vec<_> = Planner::new(FuzzSettings::default()) + .lower(&w) + .unwrap() + .into_iter() + .enumerate() + .collect(); + let initial = LogicalMemo::from_roots(&roots); + let inventory = generate_per_computation_candidates(roots).unwrap(); + let memo = try_populate_pass1(initial, &inventory).unwrap(); + let expanded = + expand_concrete_summary_state_groups_in_domain(memo, PaneWidthDomain::SeedGrids) + .unwrap(); + let (owner, candidate) = expanded + .memo + .candidates + .iter() + .enumerate() + .flat_map(|(owner, candidates)| candidates.iter().map(move |c| (owner, c))) + .find(|(_, c)| { + c.origin == CandidateOrigin::Rollup + && asap_types::ir::OperatorNode::reachable(&c.node) + .iter() + .any(|node| { + matches!(node.asap(), Some(ASAPOp::FinalizeExactAccumulator { child }) + if matches!(child.asap(), Some(ASAPOp::SummaryMerge { group_by:Some(_), .. }))) + }) + }) + .expect("integer count state rollup candidate"); + let rows = vec![ + vec![ + Value::Timestamp(0), + Value::Utf8("NYC".into()), + Value::Utf8("a".into()), + Value::Null, + ], + vec![ + Value::Timestamp(0), + Value::Utf8("NYC".into()), + Value::Utf8("b".into()), + Value::Float64(10.0), + ], + vec![ + Value::Timestamp(0), + Value::Null, + Value::Utf8("a".into()), + Value::Null, + ], + ]; + let before = + physical_common::execute_raw_rows(&expanded.memo.groups[owner].node, rows.clone()); + let after = physical_common::execute_raw_rows(&candidate.node, rows); + let sorted = |rows: Vec>| { + let mut rows: Vec<_> = rows.iter().map(|row| format!("{row:?}")).collect(); + rows.sort(); + rows + }; + assert!(after + .iter() + .all(|row| matches!(row.last(), Some(Value::Int64(_))))); + assert_eq!(sorted(before), sorted(after)); + let before_empty = + physical_common::execute_raw_rows(&expanded.memo.groups[owner].node, vec![]); + let after_empty = physical_common::execute_raw_rows(&candidate.node, vec![]); + assert_eq!(sorted(before_empty), sorted(after_empty)); + } +} + +/// #922: duplicate exact TopK readers share their result and inner window work. +#[test] +fn duplicate_exact_topk_shares_work() { + let mut w = workload(924); + for query in &mut w.queries { + query.entry.query.0 = "topk(3, sum_over_time(m_gauge[2m]))".into(); + } + let mut base = w.clone(); + base.queries.truncate(1); + let single = cost(&base); + let duplicate = cost(&w); + assert!(duplicate <= single * 1.001, "{duplicate} vs {single}"); +} + +/// #924/#925 SQL variants: nested windows and slower readers reuse one sum. +#[test] +fn sql_sum_shares_nested_windows_and_slower_readers() { + let mut w = workload(922); + w.queries.truncate(1); + w.queries[0].entry.query.0 = + "SELECT SUM(value) FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE".into(); + w.queries[0].entry.demand = serde_json::from_value(json!({"fixed_interval":60000})).unwrap(); + let single = cost(&w); + let mut nested = w.clone(); + let mut shorter = w.queries[0].clone(); + shorter.entry.query.0 = shorter.entry.query.0.replace("'5'", "'1'"); + nested.queries.push(shorter); + let nested_cost = cost(&nested); + assert!(nested_cost <= single * 1.001, "{nested_cost} vs {single}"); + let mut slower = w.queries[0].clone(); + slower.entry.demand = serde_json::from_value(json!({"fixed_interval":120000})).unwrap(); + w.queries.push(slower); + let repeated_cost = cost(&w); + assert!( + repeated_cost <= single * 1.001, + "{repeated_cost} vs {single}" + ); +} + +/// #923: sketches over an inner range function expose the same finer rollup. +#[test] +fn sketch_rollup_crosses_inner_range_function() { + use asap_logical_asap_optimization::{ + memo::{ + candidates::{try_populate_pass1, CandidateOrigin}, + state_configurations::PaneWidthDomain, + state_groups::expand_concrete_summary_state_groups_in_domain, + LogicalMemo, + }, + per_computation_candidates::generate_per_computation_candidates, + }; + let mut w = workload(923); + for (query, keys) in w.queries.iter_mut().zip(["job, region", "job"]) { + query.entry.query.0 = format!("quantile by ({keys}) (0.95, sum_over_time(m_gauge[5m]))"); + query.entry.requirements.accuracy = + asap_types::ir::properties::AccuracyTarget::max_error(0.05).failure_probability(0.01); + } + let roots: Vec<_> = Planner::new(FuzzSettings::default()) + .lower(&w) + .unwrap() + .into_iter() + .enumerate() + .collect(); + let initial = LogicalMemo::from_roots(&roots); + let inventory = generate_per_computation_candidates(roots).unwrap(); + let memo = try_populate_pass1(initial, &inventory).unwrap(); + let expanded = + expand_concrete_summary_state_groups_in_domain(memo, PaneWidthDomain::SeedGrids).unwrap(); + let owner = expanded + .memo + .groups + .iter() + .position(|g| g.roots_of.contains(&1)) + .unwrap(); + assert!(expanded.memo.candidates[owner] + .iter() + .any(|c| c.origin == CandidateOrigin::Rollup)); +} + +/// #886: unsupported per-group SQL count sketches are explicitly pruned; +/// the exact counter and separately typed raw-row TopK fusion remain available. +#[test] +fn sql_grouped_count_sketches_are_not_offered() { + use asap_logical_asap_optimization::per_computation_candidates::{ + generate_per_computation_candidates, ComputationCandidate, PruneReason, + }; + use asap_types::ir::{operator::AggIntent, NonASAPOp}; + let w = workload(886); + let roots = Planner::new(FuzzSettings::default()).lower(&w).unwrap(); + let inventory = + generate_per_computation_candidates(roots.into_iter().enumerate().collect()).unwrap(); + let target = inventory.targets.iter().find(|target| matches!(target.aggregate.non_asap(), Some(NonASAPOp::Aggregate { measures, .. }) if matches!(measures.as_slice(), [AggIntent::Count { .. }]))).unwrap(); + assert!(target + .choices + .candidates + .iter() + .all(|candidate| !matches!(candidate, ComputationCandidate::Summary(_)))); + assert_eq!( + target + .choices + .pruned + .iter() + .filter(|(_, reason)| *reason == PruneReason::NoRelationalCountSketchRealization) + .count(), + 2 + ); + assert!(inventory + .targets + .iter() + .any(|target| target.fused.is_some())); +} From acdccb864f710988da1ba05b7bae930cf954d5be Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 03:14:54 +0000 Subject: [PATCH 09/11] test(optimizer): isolate wall-clock planning budgets in sharing regressions --- crates/integration-tests/tests/identical_exact_queries.rs | 4 ++++ crates/integration-tests/tests/sharing_issue_regressions.rs | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/crates/integration-tests/tests/identical_exact_queries.rs b/crates/integration-tests/tests/identical_exact_queries.rs index d25f04f12..490fa2aed 100644 --- a/crates/integration-tests/tests/identical_exact_queries.rs +++ b/crates/integration-tests/tests/identical_exact_queries.rs @@ -21,6 +21,10 @@ fn workload() -> FuzzWorkload { } fn cost(w: &FuzzWorkload) -> f64 { + // Each regression has a wall-clock planning budget. Concurrent solver + // runs in this test binary must not consume one another's CPU budget. + static PLANNING: std::sync::Mutex<()> = std::sync::Mutex::new(()); + let _planning = PLANNING.lock().unwrap(); let roots = Planner::new(FuzzSettings::default()).lower(w).unwrap(); let planning = w.planning_workload(); let entries: Vec<_> = planning.query_workload.entries().collect(); diff --git a/crates/integration-tests/tests/sharing_issue_regressions.rs b/crates/integration-tests/tests/sharing_issue_regressions.rs index 91ef2cfa7..7e1ca2084 100644 --- a/crates/integration-tests/tests/sharing_issue_regressions.rs +++ b/crates/integration-tests/tests/sharing_issue_regressions.rs @@ -32,6 +32,10 @@ fn workload(issue: u32) -> FuzzWorkload { } fn cost(w: &FuzzWorkload) -> f64 { + // Each regression has a wall-clock planning budget. Concurrent solver + // runs in this test binary must not consume one another's CPU budget. + static PLANNING: std::sync::Mutex<()> = std::sync::Mutex::new(()); + let _planning = PLANNING.lock().unwrap(); let roots = Planner::new(FuzzSettings::default()).lower(w).unwrap(); let planning = w.planning_workload(); let entries: Vec<_> = planning.query_workload.entries().collect(); From 3d634d347b5ccaf33a950bfdd462b4c4a5bc22c4 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 03:45:54 +0000 Subject: [PATCH 10/11] test(optimizer): exercise filter subquery choices with supported quantiles --- .../tests/stage1_logical_asap_optimization.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs index 75d7a3922..ec7f584a9 100644 --- a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs +++ b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs @@ -256,11 +256,11 @@ fn topk_over_an_aggregate_ranks_its_labels() { } /// A subquery inside a measure filter is substituted too: each choice for -/// the inner count gives its own DAG, and an approximate inner count clears +/// the inner quantile gives its own DAG, and an approximate inner result clears /// the outer measure's guarantee. #[tokio::test] async fn subquery_in_a_measure_filter_is_substituted() { - let query = "SELECT src, count(*) FILTER (WHERE bytes > (SELECT count(*) FROM flows)) \ + let query = "SELECT src, count(*) FILTER (WHERE bytes > (SELECT approx_percentile_cont(bytes, 0.5) FROM flows)) \ FROM flows GROUP BY src"; let inventory = generate_per_computation_candidates(vec![(0, sql(query).await)]).unwrap(); assert_eq!(inventory.targets.len(), 2); @@ -277,7 +277,7 @@ async fn subquery_in_a_measure_filter_is_substituted() { }) .is_some_and(|by| by.keys().is_empty()) }) - .expect("inner count"); + .expect("inner quantile"); let candidates = &inventory.targets[inner].choices.candidates; assert!(candidates.len() > 1); let (mut seen, mut approximate_inner) = (Vec::new(), false); From ede27c66cc629ac58c09e20cdac8f08845004c53 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 27 Oct 2026 04:21:37 +0000 Subject: [PATCH 11/11] test(optimizer): align SQL candidate coverage with executable families --- .../tests/per_computation_candidates.rs | 146 +++++++++++------- 1 file changed, 90 insertions(+), 56 deletions(-) diff --git a/crates/logical-asap-optimization/tests/per_computation_candidates.rs b/crates/logical-asap-optimization/tests/per_computation_candidates.rs index 54e0226f3..38dda4e88 100644 --- a/crates/logical-asap-optimization/tests/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/tests/per_computation_candidates.rs @@ -404,10 +404,7 @@ fn multi_measure_aggregate_has_one_target_per_measure() { candidates[0].choices.candidates, vec![ComputationCandidate::ExactAccumulator(vec![ExactKind::Sum])] ); - assert_eq!( - algorithms(&candidates[1].choices.candidates), - vec![SketchAlgorithm::Cms, SketchAlgorithm::UnivMon] - ); + assert_sql_count_choices(&candidates[1].choices); } /// An aggregate with no realization besides exact execution is not a target. @@ -425,7 +422,7 @@ fn aggregate_without_realization_is_not_a_target() { ); } -/// A filtered measure keeps its accumulator and summaries and records its +/// A filtered SQL count keeps its executable accumulator and records its /// filter, which composition applies to the state. #[test] fn filtered_measure_keeps_candidates_and_records_its_filter() { @@ -443,19 +440,54 @@ fn filtered_measure_keeps_candidates_and_records_its_filter() { .targets; assert_eq!(targets.len(), 2); for target in &targets { - assert_eq!( - target.choices.candidates, - candidates_for(count()), - "measure {}", - target.measure - ); - assert!(target.choices.pruned.is_empty()); + assert_sql_count_choices(&target.choices); assert!(target.having.is_empty()); } assert_eq!(targets[0].filter, Some(always())); assert_eq!(targets[1].filter, None); } +fn assert_sql_count_choices( + choices: &asap_logical_asap_optimization::per_computation_candidates::ComputationChoices, +) { + assert_eq!( + choices.candidates, + vec![ComputationCandidate::ExactAccumulator(vec![ + ExactKind::Count + ])] + ); + assert_eq!( + choices.pruned, + [SketchAlgorithm::Cms, SketchAlgorithm::UnivMon] + .into_iter() + .map(|algorithm| ( + ComputationCandidate::Summary(algorithm), + PruneReason::NoRelationalCountSketchRealization, + )) + .collect::>() + ); +} + +fn quantile() -> AggIntent { + AggIntent::Quantile { + col: Some(1), + q: 0.5, + accuracy: approximate(), + } +} + +fn grouped_quantile(having: Option) -> Rc { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + child: scan(false), + reduction: Reduction::by(vec![0]), + measures: vec![quantile()], + output_names: vec![], + filters: vec![], + having, + })) + .unwrap() +} + /// `count(*) by (src_ip)`, the form the SQL frontend gives `GROUP BY src_ip`. fn grouped_count(having: Option) -> Rc { OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { @@ -485,7 +517,7 @@ fn equals(column: usize, value: ScalarValue) -> Predicate { } /// A `HAVING` over group keys only is pushed below the aggregate and keeps -/// every candidate. +/// every executable candidate. #[test] fn key_only_having_is_pushed_below_and_keeps_candidates() { let node = grouped_count(None); @@ -503,17 +535,11 @@ fn key_only_having_is_pushed_below_and_keeps_candidates() { placement: HavingPlacement::BelowAggregate, }] ); - assert_eq!( - targets[0].choices.candidates, - candidates_for(AggIntent::Count { - accuracy: approximate() - }) - ); - assert!(targets[0].choices.pruned.is_empty()); + assert_sql_count_choices(&targets[0].choices); } /// A `HAVING` over an aggregate value, as a `Filter` above or as the -/// aggregate's own `having`, stays above the output and keeps every candidate. +/// aggregate's own `having`, stays above the output and keeps executable candidates. #[test] fn value_having_stays_above_output_and_keeps_candidates() { let value = || equals(1, ScalarValue::Int64(10)); @@ -530,13 +556,7 @@ fn value_having_stays_above_output_and_keeps_candidates() { for target in &targets { let placements: Vec<_> = target.having.iter().map(|entry| entry.placement).collect(); assert_eq!(placements, vec![HavingPlacement::AboveOutput]); - assert_eq!( - target.choices.candidates, - candidates_for(AggIntent::Count { - accuracy: approximate() - }) - ); - assert!(target.choices.pruned.is_empty()); + assert_sql_count_choices(&target.choices); } } @@ -964,7 +984,20 @@ fn columns(node: &OperatorNode) -> Vec<(String, FieldDataType, bool)> { .map(|field| (field.name.clone(), field.dtype.clone(), field.nullable)) .collect() } +// Quantile readouts restore the aggregate's output names with an identity projection. +fn without_output_aliases(node: &OperatorNode) -> &OperatorNode { + if let Some(NonASAPOp::Project { cols, child, .. }) = node.non_asap() { + assert!(cols + .iter() + .enumerate() + .all(|(index, item)| item.expr == ScalarExpr::Column(index))); + without_output_aliases(child) + } else { + node + } +} fn summary_agg(node: &OperatorNode) -> &ASAPOp { + let node = without_output_aliases(node); match node.asap() { Some(ASAPOp::SummaryEstimate { summary_input, .. }) => summary_agg(summary_input), Some(ASAPOp::FinalizeExactAccumulator { child }) => summary_agg(child), @@ -978,9 +1011,7 @@ fn summary_agg(node: &OperatorNode) -> &ASAPOp { #[test] fn composed_summary_is_sized_for_its_measure() { use asap_logical_asap_optimization::realization::default_size_params; - let intent = AggIntent::Count { - accuracy: approximate(), - }; + let intent = quantile(); let [root] = <[_; 1]>::try_from(composed(vec![aggregate(intent.clone())], &[1])).unwrap(); let ASAPOp::SummaryAgg { family: FieldDataType::Sketch(kind, _), @@ -989,10 +1020,10 @@ fn composed_summary_is_sized_for_its_measure() { else { panic!("sketch state expected") }; - assert_eq!(*kind.algorithm(), SketchAlgorithm::Cms); + assert_eq!(*kind.algorithm(), SketchAlgorithm::Kll); assert_eq!( kind.params(), - &default_size_params(SketchAlgorithm::Cms, &intent, 0.05, 0.01) + &default_size_params(SketchAlgorithm::Kll, &intent, 0.05, 0.01) ); } @@ -1001,19 +1032,17 @@ fn composed_summary_is_sized_for_its_measure() { /// above the estimate. #[test] fn composed_filters_and_having() { - let cms = 1; + let kll = 1; let filtered = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { child: scan(false), reduction: Reduction::by(vec![0]), - measures: vec![AggIntent::Count { - accuracy: approximate(), - }], + measures: vec![quantile()], output_names: vec![], filters: vec![Some(always())], having: None, })) .unwrap(); - let [root] = <[_; 1]>::try_from(composed(vec![filtered.clone()], &[cms])).unwrap(); + let [root] = <[_; 1]>::try_from(composed(vec![filtered.clone()], &[kll])).unwrap(); assert!( matches!(summary_agg(&root), ASAPOp::SummaryAgg { filter: Some(f), child, .. } if *f == always() && child.non_asap().is_some_and(|op| matches!(op, NonASAPOp::Scan { .. }))) @@ -1022,8 +1051,8 @@ fn composed_filters_and_having() { let key = equals(0, ScalarValue::Utf8("a".into())); let [root] = <[_; 1]>::try_from(composed( - vec![having_filter(grouped_count(None), key.clone())], - &[cms], + vec![having_filter(grouped_quantile(None), key.clone())], + &[kll], )) .unwrap(); let ASAPOp::SummaryAgg { child, filter, .. } = summary_agg(&root) else { @@ -1035,16 +1064,19 @@ fn composed_filters_and_having() { let value = || equals(1, ScalarValue::Int64(10)); for root in composed( vec![ - having_filter(grouped_count(None), value()), - grouped_count(Some(value())), + having_filter(grouped_quantile(None), value()), + grouped_quantile(Some(value())), ], - &[cms, cms], + &[kll, kll], ) { let Some(NonASAPOp::Filter { pred, child }) = root.non_asap() else { panic!("value HAVING above the estimate expected, got {root:#?}") }; assert_eq!(*pred, value()); - assert!(matches!(child.asap(), Some(ASAPOp::SummaryEstimate { .. }))); + assert!(matches!( + without_output_aliases(child).asap(), + Some(ASAPOp::SummaryEstimate { .. }) + )); } } @@ -1086,9 +1118,7 @@ fn composed_multi_measure_aggregate_joins_its_measures() { reduction, measures: vec![ AggIntent::Sum { col: Some(1) }, - AggIntent::Count { - accuracy: approximate(), - }, + quantile(), AggIntent::StdDev { col: Some(1), population: true, @@ -1173,7 +1203,7 @@ fn composed_guarantees() { accuracy: approximate(), }), aggregate(AggIntent::Sum { col: Some(1) }), - having_filter(grouped_count(None), value), + having_filter(grouped_quantile(None), value), ], &[1, 0, 1], ); @@ -1193,25 +1223,29 @@ fn composed_guarantees() { #[test] fn own_having_subquery_is_substituted() { use asap_logical_asap_optimization::per_computation_candidates::compose_logical_candidate; - let inner = aggregate(AggIntent::Count { - accuracy: approximate(), - }); - let outer = grouped_count(Some(above_scalar(1, inner.clone()))); + let inner = aggregate(quantile()); + let outer = grouped_quantile(Some(above_scalar(1, inner.clone()))); let inventory = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(outer.clone()))]).unwrap(); assert_eq!(inventory.targets.len(), 2); - let cms = 1; - let composed = compose_logical_candidate(&inventory, &[cms, cms]).unwrap(); + let kll = 1; + let composed = compose_logical_candidate(&inventory, &[kll, kll]).unwrap(); let QueryRoot::Operator(root) = &composed[0].1 else { panic!("operator root expected") }; let Some(NonASAPOp::Filter { pred, child }) = root.non_asap() else { panic!("own HAVING above the estimate expected, got {root:#?}") }; - assert!(matches!(child.asap(), Some(ASAPOp::SummaryEstimate { .. }))); + assert!(matches!( + without_output_aliases(child).asap(), + Some(ASAPOp::SummaryEstimate { .. }) + )); let [subquery] = <[_; 1]>::try_from(pred.0.operator_refs()).unwrap(); assert!( - matches!(subquery.asap(), Some(ASAPOp::SummaryEstimate { .. })), + matches!( + without_output_aliases(subquery).asap(), + Some(ASAPOp::SummaryEstimate { .. }) + ), "{subquery:#?}" ); assert!(OperatorNode::reachable(root)