From db610540f2e2b7190eef3118e85603ddc3524ca9 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 24 Aug 2026 08:07:49 -0600 Subject: [PATCH 1/5] feat(asap-aware-mapping): rebuild rollup.rs fresh on top of #251's current tip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This branch had been forked at the very first draft of #251 (boundary::implementation_for_with, a ForceSketchKind CostModel-wrapping hack to steer bind::implement_tree_with) and never rebased — missing the "make binding literally a selector over ReplacementStrategy" refactor, and the deletion of bind::implement_tree/implement_tree_with. It would not compile against the current tip of feat/replacement-strategy-251. Its own copy of replacement.rs/lib.rs was just a frozen snapshot of that first draft, reintroducing already-fixed stale code. Rebuilt fresh: reset this branch onto the current #259 tip (which already carries the current, correct replacement.rs/bind.rs/lib.rs/ implementation.rs/cost_model.rs), then reapplied only the genuinely new content this branch adds on top of it — rollup.rs (issue #254, part of - rollup.rs itself needed no code changes at all: it never called bind::implement_tree/implement_tree_with or referenced boundary:: directly — RollupStrategy/is_legal_rollup_source/build_rollup are self-contained around crate::replacement's TargetSubDAG/ ReplacementSubDAG/Replacement/ReplacementStrategy vocabulary, which is unchanged in shape on the current #251 tip. Copied over verbatim. - lib.rs: added `pub mod rollup;` and a module-doc bullet under `## Status` describing RollupStrategy/is_legal_rollup_source (the old branch's own lib.rs diff added no such bullet and no public re-exports for rollup, so none were reintroduced here either — the module's own `pub struct`/`pub fn` items are reachable via `asap_aware_mapping::rollup::*` same as before). - Did not touch replacement.rs/bind.rs/cost_model.rs/implementation.rs: they already reflect #259's current, correct state. Verified: cargo build --workspace --all-targets, cargo test --workspace (0 failures, including all 16 rollup:: tests unchanged), cargo fmt --all -- --check, cargo clippy --workspace --all-targets --all-features -- -D warnings — all clean. Co-Authored-By: Claude Sonnet 5 --- crates/asap-aware-mapping/src/lib.rs | 12 + crates/asap-aware-mapping/src/rollup.rs | 751 ++++++++++++++++++++++++ 2 files changed, 763 insertions(+) create mode 100644 crates/asap-aware-mapping/src/rollup.rs diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index ac4ce3eb..4e41f489 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -97,6 +97,17 @@ //! #247's own rule-based traversal, which re-walked the tree once per //! optimization before [`replacement::search_workload`] existed to read //! from instead — see that module's docs for the full reframing. +//! - [`rollup`] — [`rollup::RollupStrategy`] wraps group-by-lattice roll-up +//! reuse (issue #254, part of #33) as a [`ReplacementStrategy`]: given a +//! coarser `Aggregate` target and a caller-supplied sibling set, proposes +//! re-deriving it from an already-computed, strictly finer sibling +//! `Aggregate` over the same shared child instead of an independent pass +//! over the raw source — the cross-aggregate sibling of +//! `pre_asap::cse::share_common_subtrees`'s identical-subtree sharing. +//! [`rollup::is_legal_rollup_source`] is the standalone legality predicate +//! other axes (e.g. issue #256's `GroupingStrategy`) are expected to +//! consult directly, so it and this module's `RollupStrategy` can never +//! disagree about which siblings qualify. //! //! ## Terminology — "bind" already means three different things nearby; //! this crate's own logical→physical step is named "implementation" instead @@ -141,6 +152,7 @@ pub mod cost_model; pub mod explanation; pub mod replacement; +pub mod rollup; pub use cost_model::{CostModel, DefaultCostModel}; pub use explanation::{ diff --git a/crates/asap-aware-mapping/src/rollup.rs b/crates/asap-aware-mapping/src/rollup.rs new file mode 100644 index 00000000..5f5520ca --- /dev/null +++ b/crates/asap-aware-mapping/src/rollup.rs @@ -0,0 +1,751 @@ +//! [`RollupStrategy`] — group-by-lattice roll-up reuse (fine-to-coarse) as a +//! [`ReplacementStrategy`] (issue #254, part of #33). +//! +//! ## The optimization: AHA-style roll-up, not independent subpopulations +//! +//! `docs/asap_aware_mapping.md`'s "Degrees of freedom" section names this +//! axis directly: **"AHA vs treating hierarchical subpopulations +//! independently."** Two `Aggregate` nodes over the same source, grouped at +//! different granularities (e.g. `by (job)` and `by (job, region)`), are +//! today two *independent* passes over the source. But when the coarser +//! grouping's keys are a subset of the finer grouping's keys, the coarser +//! answer is derivable from the finer one *without ever touching the raw +//! source again* — the same "roll up the group-by lattice" reuse a SQL +//! `ROLLUP`/OLAP-cube engine already exploits between grouping levels of one +//! `GROUP BY ROLLUP(...)` (see `frontend-sql`'s own `ROLLUP` lowering +//! tests), generalized here to two *independently written* aggregates in a +//! workload rather than one multi-level `ROLLUP` clause. +//! +//! This is exactly Peilin's catalog entry (issue #33's comment thread): +//! "Rolling up aggregations on a fine-grained group by to get a +//! coarse-grained group by (like AHA)," alongside "CSE across aggregations, +//! and group by key management" — this strategy is the *cross-aggregate* +//! sibling of `pre_asap::cse::share_common_subtrees`'s *identical*-subtree +//! sharing: CSE shares two structurally-*equal* aggregates onto one `Rc`; +//! this strategy relates two structurally-*different* (differently grouped) +//! aggregates over the same shared source. +//! +//! ## Why re-aggregating the finer side needs a *combinator*, not literally +//! "the same measure" reapplied +//! +//! Re-deriving `by (job)` from `by (job, region)` means computing, for each +//! `job`, "the combination of every `(job, region)` partial result for that +//! `job`." Whether that combination is *the same operator reapplied* depends +//! on the operator: +//! +//! - `Sum`/`Min`/`Max` are **self-combining**: a sum of sums is a sum, a min +//! of mins is a min. Reapplying the identical `AggIntent` over the finer +//! side's own output column is correct. +//! - `Count` is **not** self-combining: re-`Count`ing the finer side's own +//! output rows counts the number of *finer groups* per coarser group (how +//! many distinct `region`s a `job` has), not the original row count. The +//! correct combinator is `Sum` — `count(A ∪ B) = count(A) + count(B)`, the +//! same "`COUNT(*)` over a pre-aggregated summary table becomes +//! `SUM(count)`" rule every OLAP rollup/cube/materialized-view-matching +//! implementation applies. `Increase` mirrors it (a counter's total +//! increase across sub-windows is additive), so it combines via `Sum` too. +//! - `Rate` (`increase / duration`) has **no** valid self- or sum-combinator +//! here, so it is deliberately left unhandled (see [`rollup_combinator`]). +//! In practice this never matters: `Rate`/`Increase` are constructed with +//! `Reduction::PerEntity` (per-series, no `by` at all — see +//! `AggIntent::is_per_series` and `asap_frontend_promql::promql::reduction_for`), +//! never `Reduction::Reduce`, so they never carry a `by` set for this +//! strategy to compare in the first place; `PerEntity` nodes never match +//! ([`bindable_grouped_aggregate`] returns `None` for them). `Increase`'s +//! entry in [`rollup_combinator`] is a defensive completion of the match +//! (exhaustive-over-the-mergeable-vocabulary, "no silent fallthrough"), +//! not a case expected to fire today. +//! +//! [`rollup_combinator`] is the single place this substitution is decided — +//! [`is_legal_rollup_source`] (the standalone legality predicate issue #254 +//! asks for) consults it, and so does this module's `replacements`, so the +//! two can never disagree about which intents are eligible. +//! +//! ## `ColumnId` comparability — only sound because the child `Rc` is shared +//! +//! A `ColumnId` is a *position* into a specific `Schema` (`crates/types/src/pre_asap/schema.rs`'s +//! own doc: "the same edge, the same schema, the same positional numbering"). +//! Comparing the coarser aggregate's `by` positions against the finer +//! aggregate's `by` positions is only meaningful because **both aggregates +//! share the identical child `Rc`** (post-CSE, per +//! `pre_asap::cse::share_common_subtrees`'s doc on why hash-consing is a +//! precondition, not a coincidence) — so both `by` lists are positional +//! offsets into the exact same schema. Two *structurally different but +//! equivalent* sources (e.g. two independently-built scans of the same +//! underlying table with columns in a different order) are explicitly out +//! of scope: this module never attempts to reconcile `ColumnId`s across two +//! distinct schemas. +//! +//! ## Non-goals (tracked separately, not attempted here — same split +//! `replacement.rs`'s own module docs draw for `SharedSubtreeStrategy`'s +//! `consumer_count`) +//! +//! - **No workload-wide sibling discovery.** Finding "every `Aggregate` node +//! across a whole workload that shares a given `Rc` child" is a traversal +//! over the *whole* workload, not a fact available from one +//! [`TargetSubDAG`] in isolation — [`ReplacementStrategy::matches`]/ +//! `replacements` only ever see one target at a time. [`RollupStrategy`] +//! takes the already-discovered sibling set as constructor state instead +//! of rediscovering it: a future caller (the search engine, issue #252, +//! built in parallel and possibly not yet landed on this branch) is +//! expected to construct this strategy with the real sibling set it +//! already found — the same "this module wraps a decision, it does not +//! own the traversal that feeds it" split `replacement.rs` draws for +//! `SharedSubtreeStrategy::matches`'s `consumer_count`. +//! - **No materialized roll-up operator.** Actually building a pre-aggregated +//! summary/scan leaf at execution time is separate, larger work outside +//! `asap-aware-mapping`'s scope (see issue #254's own "Non-goal" section) +//! — this module only constructs the pre-ASAP [`QueryExpr::Aggregate`] +//! rewrite; a `CostModel`/search engine decides whether to prefer it. +//! - **No cross-schema reconciliation** (see "`ColumnId` comparability" +//! above) and **no `without(...)` grouping support** — `without`'s kept +//! set is runtime-open (never enumerable at plan time, per +//! `GroupKeys`'s own doc), so there is no fixed `ColumnId` set to compare +//! against a superset/subset relationship at all; [`is_legal_rollup_source`] +//! declines both directions. + +use std::collections::HashSet; +use std::rc::Rc; + +use asap_types::pre_asap::agg_intent::AggIntent; +use asap_types::pre_asap::query_expr::{GroupKeys, QueryExpr, Reduction}; +use asap_types::pre_asap::schema::{ColumnId, Schema}; + +use crate::replacement::{Replacement, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG}; + +/// The `(by, intent, child)` shape this strategy operates on: a single +/// measure, no `HAVING` — the same bindable shape +/// [`crate::replacement::SketchAlgorithmStrategy`] requires (see that module's +/// private `bindable_intent`) — **plus** a genuine [`Reduction::Reduce`] +/// grouping to compare (not [`Reduction::PerEntity`], which has no `by` set +/// at all). `None` for anything else, including a multi-measure or `HAVING` +/// aggregate, a non-`Aggregate` node, or a `PerEntity` reduction. +fn bindable_grouped_aggregate( + node: &QueryExpr, +) -> Option<(&GroupKeys, &AggIntent, &Rc)> { + let QueryExpr::Aggregate { + reduction, + measures, + having, + child, + .. + } = node + else { + return None; + }; + let ([intent], None) = (measures.as_slice(), having) else { + return None; + }; + let Reduction::Reduce(by) = reduction else { + return None; + }; + Some((by, intent, child)) +} + +/// The `AggIntent` to combine `intent`'s partial results with, when +/// re-aggregating over an already-computed finer aggregate's own measure +/// column at position `finer_measure_col` — see the module docs' "Why +/// re-aggregating the finer side needs a *combinator*" section for the +/// reasoning behind each arm. `None` for any intent this module does not +/// (yet) know a correct combinator for — including every intent +/// `agg_is_mergeable` permits but this module doesn't specifically handle +/// (`Rate`, and everything outside the `Sum`/`Count`/`MinMax`/`Increase` +/// vocabulary `agg_is_mergeable`'s own doc names) — so `is_legal_rollup_source` +/// (which calls this) is *strictly narrower* than `agg_is_mergeable` alone, +/// deliberately: `agg_is_mergeable` answers "does *some* partial-state merge +/// exist", not "is self- or sum-recombination the right one," and this +/// module only ever proposes a rewrite it can construct correctly. +fn rollup_combinator(intent: &AggIntent, finer_measure_col: ColumnId) -> Option { + match intent { + // Self-combining: reapplying the identical operator over the finer + // side's own output column is correct unchanged. + AggIntent::Sum { .. } => Some(AggIntent::Sum { + col: Some(finer_measure_col), + }), + AggIntent::Min { .. } => Some(AggIntent::Min { + col: Some(finer_measure_col), + }), + AggIntent::Max { .. } => Some(AggIntent::Max { + col: Some(finer_measure_col), + }), + // Not self-combining — see the module docs. Both merge by addition + // over the finer side's own output column instead. + AggIntent::Count { .. } | AggIntent::Increase => Some(AggIntent::Sum { + col: Some(finer_measure_col), + }), + _ => None, + } +} + +/// **The standalone legality predicate issue #254 asks for**: is `finer`'s +/// grouping a legal roll-up source for `coarser`'s grouping, given that both +/// nodes compute an aggregation intent over the same shared child? +/// +/// Issue #256 (`GroupingStrategy`/Hydra axis, built in parallel from the +/// same base) is expected to call this exact function before assuming a +/// Hydra-backed aggregate composes with a roll-up — it is named and +/// exported for that reason, not left as private inline logic in `matches`/ +/// `replacements`. +/// +/// Legal iff, **in this order**: +/// +/// 1. `finer_intent == coarser_intent` — the two aggregates compute the +/// identical intent + column (verified here, not just assumed by the +/// caller, so this function is a complete, self-contained answer on its +/// own). +/// 2. [`rollup_combinator`] knows how to re-derive `finer_intent` from a +/// pre-aggregated column at all (excludes `Avg`/`StdDev`/`Variance` — +/// never `agg_is_mergeable` — and also excludes every `agg_is_mergeable` +/// intent this module doesn't specifically handle, e.g. `Rate`). +/// 3. Neither grouping is a `without(...)` exclusion grouping — `without`'s +/// kept set is runtime-open, so there is no fixed `ColumnId` set to +/// compare a superset/subset relationship against (see the module docs). +/// 4. `finer_output_schema` (the finer aggregate's own *output* schema, not +/// the shared child's) carries a provable unique key +/// ([`Schema::has_unique_key`]) — **the exact legality gate +/// `pre_asap::cse::share_common_subtrees` already applies to its own +/// sharing decisions**, reused verbatim here rather than re-invented: +/// `share_common_subtrees`'s own doc ("Legality: gated by +/// `Schema::unique_keys`") states a producer's output is only safely +/// reusable across consumers when its row identity is provably stable — +/// exactly the property re-aggregating over `finer` as if it were a +/// fresh source requires. +/// 5. `coarser_by` is a **strict, proper** subset of `finer_by` (same +/// `ColumnId`s, finer strictly more of them) — an *equal* `by` is +/// `SharedSubtreeStrategy`'s CSE-sharing question, not a roll-up, so +/// equality is deliberately excluded here, not treated as a degenerate +/// roll-up. +pub fn is_legal_rollup_source( + finer_by: &GroupKeys, + finer_output_schema: &Schema, + finer_intent: &AggIntent, + coarser_by: &GroupKeys, + coarser_intent: &AggIntent, +) -> bool { + if finer_intent != coarser_intent { + return false; + } + if rollup_combinator(finer_intent, 0).is_none() { + return false; + } + if finer_by.is_without() || coarser_by.is_without() { + return false; + } + if !finer_output_schema.has_unique_key() { + return false; + } + is_strict_column_superset(finer_by.keys(), coarser_by.keys()) +} + +/// Whether `finer` is a strict, proper superset of `coarser` — every +/// `ColumnId` in `coarser` also appears in `finer`, and `finer` has more of +/// them (an equal-length or shorter `finer` can never be a proper +/// superset, so the length check alone rules out equality without a set +/// comparison). +fn is_strict_column_superset(finer: &[ColumnId], coarser: &[ColumnId]) -> bool { + if finer.len() <= coarser.len() { + return false; + } + let finer_set: HashSet<&ColumnId> = finer.iter().collect(); + coarser.iter().all(|id| finer_set.contains(id)) +} + +/// Wraps the group-by-lattice roll-up reuse (issue #254, part of #33) as a +/// [`ReplacementStrategy`]: given a coarser `Aggregate` [`TargetSubDAG`], +/// finds every already-known sibling `Aggregate` that shares the same child +/// `Rc` and is a legal, strictly finer roll-up source for it (per +/// [`is_legal_rollup_source`]), and proposes replacing the target with a +/// re-aggregation over that sibling instead of the shared raw source. +/// +/// `siblings` is **caller-supplied, not discovered here** — see the module +/// docs' "Non-goals" on why finding the full sibling set across a workload +/// is a workload-wide traversal this strategy does not own. +pub struct RollupStrategy<'a> { + siblings: &'a [Rc], +} + +impl<'a> RollupStrategy<'a> { + /// A strategy that considers every node in `siblings` as a candidate + /// roll-up source (or target) — typically the full set of `Aggregate` + /// nodes a workload-wide discovery pass (issue #252) already found + /// sharing at least one child `Rc` with something else. + pub fn new(siblings: &'a [Rc]) -> Self { + Self { siblings } + } + + /// Every sibling that is a legal, strictly finer roll-up source for + /// `target` — shared between `matches` and `replacements` so the two + /// can never disagree about which siblings qualify. + fn finer_sources(&self, target: &TargetSubDAG<'_>) -> Vec<&'a Rc> { + let Some((coarser_by, coarser_intent, coarser_child)) = + bindable_grouped_aggregate(target.root) + else { + return Vec::new(); + }; + + self.siblings + .iter() + .filter(|&candidate| { + if Rc::ptr_eq(candidate, target.root) { + return false; + } + let Some((finer_by, finer_intent, finer_child)) = + bindable_grouped_aggregate(candidate) + else { + return false; + }; + if !Rc::ptr_eq(finer_child, coarser_child) { + return false; + } + let Ok(finer_schema) = candidate.output_schema() else { + return false; + }; + is_legal_rollup_source( + finer_by, + &finer_schema, + finer_intent, + coarser_by, + coarser_intent, + ) + }) + .collect() + } +} + +impl ReplacementStrategy for RollupStrategy<'_> { + fn matches(&self, target: &TargetSubDAG<'_>) -> bool { + !self.finer_sources(target).is_empty() + } + + fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec { + let Some((coarser_by, coarser_intent, _)) = bindable_grouped_aggregate(target.root) else { + return Vec::new(); + }; + self.finer_sources(target) + .into_iter() + .filter_map(|finer| build_rollup(finer, coarser_by, coarser_intent)) + .collect() + } +} + +/// Build the coarser replacement: a new `QueryExpr::Aggregate` grouped by +/// `coarser_by`'s columns (repositioned into `finer`'s own output schema — +/// see below), computing `rollup_combinator(intent, ..)` over `finer`'s own +/// measure column, with `child = finer` instead of the original shared +/// source. +/// +/// `coarser_by`'s `ColumnId`s are positions into the *shared child's* +/// schema (the same schema `finer_by`'s `ColumnId`s index into — see the +/// module docs' "`ColumnId` comparability" section). `finer`'s own output +/// schema is a *different* schema (`finer_by`'s columns, in order, followed +/// by its one measure column — `aggregate_output_schema`'s `by ++ measures` +/// shape), so each of `coarser_by`'s columns must be translated from its +/// position in the shared child to its position in `finer`'s output: the +/// index its `ColumnId` occupies within `finer_by`'s own ordered list. +fn build_rollup( + finer: &Rc, + coarser_by: &GroupKeys, + intent: &AggIntent, +) -> Option { + let (finer_by, _, _) = bindable_grouped_aggregate(finer)?; + // `finer`'s own single measure sits right after its `by` columns in its + // output schema (`aggregate_output_schema`'s `by ++ measures` layout). + let finer_measure_col: ColumnId = finer_by.len(); + let combinator = rollup_combinator(intent, finer_measure_col)?; + + let remapped_by: Vec = coarser_by + .keys() + .iter() + .map(|id| finer_by.keys().iter().position(|f| f == id)) + .collect::>>()?; + + let rewritten = QueryExpr::Aggregate { + reduction: Reduction::by(remapped_by), + measures: vec![combinator], + output_names: vec![], + having: None, + child: Rc::clone(finer), + }; + + Some(ReplacementSubDAG { + replacement: Replacement::Rewrite(Rc::new(rewritten)), + rationale: format!( + "rolls up from the finer Aggregate grouped by {:?} (a strict superset of this \ + node's own {:?} grouping over the same shared source) instead of an independent \ + pass over the raw source — both compute {intent:?} over the same input, and the \ + finer side has a provable unique key (Schema::has_unique_key)", + finer_by.keys(), + coarser_by.keys(), + ), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::pre_asap::query_expr::Source; + use asap_types::pre_asap::schema::{Column, DataType}; + use asap_types::types::AccuracyTarget; + + /// `[ts(0), value(1), job(2), region(3)]`. + fn metric_scan() -> QueryExpr { + QueryExpr::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Column::new("ts", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + Column::new("job", DataType::Utf8, true), + Column::new("region", DataType::Utf8, true), + ], + 0, + vec![], + ), + } + } + + fn agg(by: Vec, intent: AggIntent, child: &Rc) -> Rc { + Rc::new(QueryExpr::Aggregate { + reduction: Reduction::by(by), + measures: vec![intent], + output_names: vec![], + having: None, + child: Rc::clone(child), + }) + } + + fn without_agg( + excluded: Vec, + intent: AggIntent, + child: &Rc, + ) -> Rc { + Rc::new(QueryExpr::Aggregate { + reduction: Reduction::Reduce(GroupKeys::without(excluded)), + measures: vec![intent], + output_names: vec![], + having: None, + child: Rc::clone(child), + }) + } + + // ── is_legal_rollup_source (the standalone predicate) ─────────────── + + #[test] + fn predicate_accepts_a_strict_superset_over_a_mergeable_intent_with_a_unique_key() { + let finer_schema = Schema::with_time_index(vec![], 0, vec![vec![0, 1]]); + assert!(is_legal_rollup_source( + &GroupKeys::by(vec![2, 3]), + &finer_schema, + &AggIntent::Sum { col: None }, + &GroupKeys::by(vec![2]), + &AggIntent::Sum { col: None }, + )); + } + + #[test] + fn predicate_rejects_mismatched_intents() { + let finer_schema = Schema::with_time_index(vec![], 0, vec![vec![0, 1]]); + assert!(!is_legal_rollup_source( + &GroupKeys::by(vec![2, 3]), + &finer_schema, + &AggIntent::Sum { col: None }, + &GroupKeys::by(vec![2]), + &AggIntent::Sum { col: Some(9) }, + )); + } + + #[test] + fn predicate_rejects_a_non_mergeable_intent() { + let finer_schema = Schema::with_time_index(vec![], 0, vec![vec![0, 1]]); + assert!(!is_legal_rollup_source( + &GroupKeys::by(vec![2, 3]), + &finer_schema, + &AggIntent::Avg { col: None }, + &GroupKeys::by(vec![2]), + &AggIntent::Avg { col: None }, + )); + } + + #[test] + fn predicate_rejects_an_intent_with_no_known_combinator() { + // Rate is `agg_is_mergeable` but this module has no correct + // self-/sum-combinator for it (see `rollup_combinator`'s doc). + let finer_schema = Schema::with_time_index(vec![], 0, vec![vec![0, 1]]); + assert!(!is_legal_rollup_source( + &GroupKeys::by(vec![2, 3]), + &finer_schema, + &AggIntent::Rate, + &GroupKeys::by(vec![2]), + &AggIntent::Rate, + )); + } + + #[test] + fn predicate_rejects_a_finer_side_with_no_unique_key() { + // A schema with an empty `unique_keys` — as if `finer` were, e.g., a + // `without(...)`-grouped or otherwise non-hoistable aggregate — even + // though the `by` sets themselves are a clean strict superset. + let no_unique_key_schema = Schema::new(vec![]); + assert!(!is_legal_rollup_source( + &GroupKeys::by(vec![2, 3]), + &no_unique_key_schema, + &AggIntent::Sum { col: None }, + &GroupKeys::by(vec![2]), + &AggIntent::Sum { col: None }, + )); + } + + #[test] + fn predicate_rejects_equal_by_sets() { + // Equality is `SharedSubtreeStrategy`'s question, not a roll-up. + let finer_schema = Schema::with_time_index(vec![], 0, vec![vec![0]]); + assert!(!is_legal_rollup_source( + &GroupKeys::by(vec![2]), + &finer_schema, + &AggIntent::Sum { col: None }, + &GroupKeys::by(vec![2]), + &AggIntent::Sum { col: None }, + )); + } + + #[test] + fn predicate_rejects_unrelated_by_sets() { + let finer_schema = Schema::with_time_index(vec![], 0, vec![vec![0, 1]]); + assert!(!is_legal_rollup_source( + &GroupKeys::by(vec![3, 4]), + &finer_schema, + &AggIntent::Sum { col: None }, + &GroupKeys::by(vec![2]), + &AggIntent::Sum { col: None }, + )); + } + + #[test] + fn predicate_rejects_a_without_grouping_on_either_side() { + let finer_schema = Schema::with_time_index(vec![], 0, vec![vec![0, 1]]); + assert!(!is_legal_rollup_source( + &GroupKeys::without(vec![2, 3]), + &finer_schema, + &AggIntent::Sum { col: None }, + &GroupKeys::by(vec![2]), + &AggIntent::Sum { col: None }, + )); + assert!(!is_legal_rollup_source( + &GroupKeys::by(vec![2, 3]), + &finer_schema, + &AggIntent::Sum { col: None }, + &GroupKeys::without(vec![2]), + &AggIntent::Sum { col: None }, + )); + } + + // ── RollupStrategy ──────────────────────────────────────────────────── + + #[test] + fn superset_by_over_identical_mergeable_intent_and_shared_child_rolls_up() { + let scan = Rc::new(metric_scan()); + let fine = agg(vec![2, 3], AggIntent::Sum { col: Some(1) }, &scan); + let coarse = agg(vec![2], AggIntent::Sum { col: Some(1) }, &scan); + + let siblings = vec![Rc::clone(&fine), Rc::clone(&coarse)]; + let strategy = RollupStrategy::new(&siblings); + let target = TargetSubDAG::new(&coarse); + + assert!(strategy.matches(&target)); + let replacements = strategy.replacements(&target); + assert_eq!(replacements.len(), 1, "{replacements:?}"); + + let Replacement::Rewrite(rewritten) = &replacements[0].replacement else { + panic!("expected a Rewrite replacement"); + }; + let QueryExpr::Aggregate { + reduction, + measures, + child, + having, + .. + } = rewritten.as_ref() + else { + panic!("expected an Aggregate rewrite, got {rewritten:?}"); + }; + assert!(having.is_none()); + assert!( + Rc::ptr_eq(child, &fine), + "child must be the finer aggregate" + ); + assert_eq!( + reduction.expect_reduce(), + &vec![0usize], + "coarser's `job` column (position 2 in the shared source) is the finer \ + aggregate's own column 0" + ); + assert_eq!( + measures, + &vec![AggIntent::Sum { col: Some(2) }], + "Sum is self-combining: re-Sum over the finer aggregate's own Sum output \ + column (position 2, right after its two `by` keys, job and region)" + ); + assert!(!replacements[0].rationale.is_empty()); + assert!(replacements[0].rationale.contains("finer")); + } + + #[test] + fn count_rolls_up_via_sum_not_count() { + // Count is not self-combining (see the module docs) — the rewritten + // measure must be Sum over the finer Count's own output column, not + // Count reapplied. + let scan = Rc::new(metric_scan()); + let fine = agg( + vec![2, 3], + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }, + &scan, + ); + let coarse = agg( + vec![2], + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }, + &scan, + ); + + let siblings = vec![Rc::clone(&fine), Rc::clone(&coarse)]; + let strategy = RollupStrategy::new(&siblings); + let target = TargetSubDAG::new(&coarse); + + let replacements = strategy.replacements(&target); + assert_eq!(replacements.len(), 1, "{replacements:?}"); + let Replacement::Rewrite(rewritten) = &replacements[0].replacement else { + panic!("expected a Rewrite replacement"); + }; + let QueryExpr::Aggregate { measures, .. } = rewritten.as_ref() else { + panic!("expected an Aggregate rewrite"); + }; + assert_eq!( + measures, + &vec![AggIntent::Sum { col: Some(2) }], + "Count's own output column sits at position 2, right after its two `by` keys" + ); + } + + #[test] + fn non_mergeable_intent_does_not_roll_up() { + let scan = Rc::new(metric_scan()); + let fine = agg(vec![2, 3], AggIntent::Avg { col: Some(1) }, &scan); + let coarse = agg(vec![2], AggIntent::Avg { col: Some(1) }, &scan); + + let siblings = vec![Rc::clone(&fine), Rc::clone(&coarse)]; + let strategy = RollupStrategy::new(&siblings); + let target = TargetSubDAG::new(&coarse); + + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } + + #[test] + fn no_unique_key_on_the_finer_side_does_not_roll_up() { + // `without(...)` groupings never carry a provable unique key + // (`without_output_schema` always reports `unique_keys: []`) even + // though its `keys()` (the *excluded* positions here) happens to be + // numerically a superset of the coarser side's *kept* positions — + // `is_legal_rollup_source` rejects any `without` grouping outright, + // and would reject on the missing unique key regardless. + let scan = Rc::new(metric_scan()); + let fine = without_agg(vec![2, 3], AggIntent::Sum { col: Some(1) }, &scan); + let coarse = agg(vec![2], AggIntent::Sum { col: Some(1) }, &scan); + + assert!( + !fine.output_schema().unwrap().has_unique_key(), + "fixture sanity: a without(...) aggregate has no provable unique key" + ); + + let siblings = vec![Rc::clone(&fine), Rc::clone(&coarse)]; + let strategy = RollupStrategy::new(&siblings); + let target = TargetSubDAG::new(&coarse); + + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } + + #[test] + fn unrelated_by_sets_do_not_roll_up() { + // Neither `[job]` nor `[region]` is a superset of the other. + let scan = Rc::new(metric_scan()); + let a = agg(vec![2], AggIntent::Sum { col: Some(1) }, &scan); + let b = agg(vec![3], AggIntent::Sum { col: Some(1) }, &scan); + + let siblings = vec![Rc::clone(&a), Rc::clone(&b)]; + let strategy = RollupStrategy::new(&siblings); + + let target_a = TargetSubDAG::new(&a); + assert!(!strategy.matches(&target_a)); + assert!(strategy.replacements(&target_a).is_empty()); + + let target_b = TargetSubDAG::new(&b); + assert!(!strategy.matches(&target_b)); + assert!(strategy.replacements(&target_b).is_empty()); + } + + #[test] + fn equal_by_sets_do_not_roll_up() { + // Equal groupings are `SharedSubtreeStrategy`'s CSE-sharing + // question (build once and share, or build independently) — a + // roll-up requires a *strict* superset, not equality. + let scan = Rc::new(metric_scan()); + let a = agg(vec![2], AggIntent::Sum { col: Some(1) }, &scan); + let b = agg(vec![2], AggIntent::Sum { col: Some(1) }, &scan); + + let siblings = vec![Rc::clone(&a), Rc::clone(&b)]; + let strategy = RollupStrategy::new(&siblings); + let target = TargetSubDAG::new(&a); + assert!(!strategy.matches(&target)); + } + + #[test] + fn different_child_does_not_roll_up_even_with_identical_shape() { + // Two separately-allocated (not CSE-shared) scans: same shape, but + // not the same `Rc`, so their `ColumnId`s are not comparable per + // this module's own scope (see the module docs). + let fine = agg( + vec![2, 3], + AggIntent::Sum { col: Some(1) }, + &Rc::new(metric_scan()), + ); + let coarse = agg( + vec![2], + AggIntent::Sum { col: Some(1) }, + &Rc::new(metric_scan()), + ); + + let siblings = vec![Rc::clone(&fine), Rc::clone(&coarse)]; + let strategy = RollupStrategy::new(&siblings); + let target = TargetSubDAG::new(&coarse); + assert!(!strategy.matches(&target)); + } + + #[test] + fn does_not_match_a_multi_measure_or_having_aggregate() { + let scan = Rc::new(metric_scan()); + let fine = agg(vec![2, 3], AggIntent::Sum { col: Some(1) }, &scan); + let multi = Rc::new(QueryExpr::Aggregate { + reduction: Reduction::by(vec![2]), + measures: vec![ + AggIntent::Sum { col: Some(1) }, + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }, + ], + output_names: vec![], + having: None, + child: Rc::clone(&scan), + }); + + let siblings = vec![Rc::clone(&fine), Rc::clone(&multi)]; + let strategy = RollupStrategy::new(&siblings); + let target = TargetSubDAG::new(&multi); + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } +} From 3dd72fbd116224d13868e1bfb2f9237c49f54bfc Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 24 Aug 2026 21:18:38 -0600 Subject: [PATCH 2/5] fix: preserve rollup output names --- crates/asap-aware-mapping/src/rollup.rs | 35 +++++++++++++++++++++++-- 1 file changed, 33 insertions(+), 2 deletions(-) diff --git a/crates/asap-aware-mapping/src/rollup.rs b/crates/asap-aware-mapping/src/rollup.rs index 5f5520ca..8f655967 100644 --- a/crates/asap-aware-mapping/src/rollup.rs +++ b/crates/asap-aware-mapping/src/rollup.rs @@ -321,9 +321,12 @@ impl ReplacementStrategy for RollupStrategy<'_> { let Some((coarser_by, coarser_intent, _)) = bindable_grouped_aggregate(target.root) else { return Vec::new(); }; + let QueryExpr::Aggregate { output_names, .. } = target.root.as_ref() else { + unreachable!("bindable_grouped_aggregate already confirmed Aggregate"); + }; self.finer_sources(target) .into_iter() - .filter_map(|finer| build_rollup(finer, coarser_by, coarser_intent)) + .filter_map(|finer| build_rollup(finer, coarser_by, coarser_intent, output_names)) .collect() } } @@ -346,6 +349,7 @@ fn build_rollup( finer: &Rc, coarser_by: &GroupKeys, intent: &AggIntent, + output_names: &[String], ) -> Option { let (finer_by, _, _) = bindable_grouped_aggregate(finer)?; // `finer`'s own single measure sits right after its `by` columns in its @@ -362,7 +366,7 @@ fn build_rollup( let rewritten = QueryExpr::Aggregate { reduction: Reduction::by(remapped_by), measures: vec![combinator], - output_names: vec![], + output_names: output_names.to_vec(), having: None, child: Rc::clone(finer), }; @@ -630,6 +634,33 @@ mod tests { ); } + #[test] + fn rollup_preserves_the_coarser_output_name() { + let scan = Rc::new(metric_scan()); + let fine = agg(vec![2, 3], AggIntent::Sum { col: Some(1) }, &scan); + let coarse = Rc::new(QueryExpr::Aggregate { + reduction: Reduction::by(vec![2]), + measures: vec![AggIntent::Sum { col: Some(1) }], + output_names: vec!["total_requests".into()], + having: None, + child: Rc::clone(&scan), + }); + let original_schema = coarse.output_schema().unwrap(); + + let siblings = vec![Rc::clone(&fine), Rc::clone(&coarse)]; + let strategy = RollupStrategy::new(&siblings); + let replacements = strategy.replacements(&TargetSubDAG::new(&coarse)); + let Replacement::Rewrite(rewritten) = &replacements[0].replacement else { + panic!("expected a Rewrite replacement"); + }; + + assert_eq!(rewritten.output_schema().unwrap(), original_schema); + let QueryExpr::Aggregate { output_names, .. } = rewritten.as_ref() else { + unreachable!(); + }; + assert_eq!(output_names, &vec!["total_requests".to_string()]); + } + #[test] fn non_mergeable_intent_does_not_roll_up() { let scan = Rc::new(metric_scan()); From 2b8789fd6567aecc33ae2e922e4a4fca4d8675e8 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 24 Aug 2026 21:33:07 -0600 Subject: [PATCH 3/5] style: format merged module declarations --- crates/asap-aware-mapping/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index f6ac26d4..d7c366db 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -166,8 +166,8 @@ pub mod cost_model; pub mod explanation; pub mod replacement; -pub mod rollup; pub mod rewrite; +pub mod rollup; pub use cost_model::{CostModel, DefaultCostModel}; pub use explanation::{ From a78cc7db278e3dc4ca8cee41aaa8e295f028efd9 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 25 Aug 2026 09:10:39 -0600 Subject: [PATCH 4/5] fix(rollup): integrate safe workload detection --- crates/asap-aware-mapping/src/replacement.rs | 38 +++- crates/asap-aware-mapping/src/rollup.rs | 189 ++++++++++++++----- 2 files changed, 174 insertions(+), 53 deletions(-) diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 3632b39b..46fef95c 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -278,6 +278,7 @@ use std::rc::Rc; use thiserror::Error; use crate::cost_model::{CostModel, CseCandidate, DefaultCostModel, ShareDecision}; +use crate::rollup::RollupStrategy; /// Errors from the pre-ASAP → post-ASAP replacement/construction path /// ([`realize_child`] and [`keep_pre_asap`]). Moved here from the former @@ -1776,9 +1777,10 @@ fn sketch_kind_of(node: &SummaryNode) -> Option { // ── default_strategies ────────────────────────────────────────────────── -/// The strategies [`search_workload`] runs, in the built-in -/// [`DefaultCostModel`] configuration — mirrors this module's own two -/// shipped [`ReplacementStrategy`] impls. +/// The context-free strategies [`search_workload`] runs in the built-in +/// [`DefaultCostModel`] configuration. Workload-dependent strategies such as +/// [`RollupStrategy`] are added by [`search_workload`] after CSE and target +/// discovery, when their sibling context exists. /// [`crate::explanation::explain_replacements`] (issue #257) uses /// this same set (via [`search_workload`]) rather than keeping a second, /// explanation-specific list to stay in sync with. Use @@ -1817,9 +1819,11 @@ pub fn search_workload(roots: Vec<(Id, Rc)>) -> PlanSpace { search_workload_with(roots, &default_strategies()) } -/// Like [`search_workload`], but with an explicit `strategies` set (see -/// [`default_strategies_with`] to plug in a deployment-specific -/// [`CostModel`] for candidate generation). +/// Like [`search_workload`], but with an explicit set of context-free +/// `strategies` (see [`default_strategies_with`] to plug in a +/// deployment-specific [`CostModel`]). The workload-dependent +/// [`RollupStrategy`] is derived and added automatically after CSE for both +/// entry points, because only this function owns the post-CSE sibling set. /// /// Runs [`share_common_subtrees`] once over `roots` first — so every /// strategy (and, transitively, every @@ -1837,6 +1841,10 @@ pub fn search_workload_with<'s, Id>( roots: Vec<(Id, Rc)>, strategies: &[Box], ) -> PlanSpace { + search_cse_workload_with(cse_workload(roots), strategies) +} + +fn cse_workload(roots: Vec<(Id, Rc)>) -> Vec<(Id, Rc)> { // `share_common_subtrees` wants owned `QueryExpr`s, not already-`Rc` // roots — the same `Rc::try_unwrap`-with-clone-fallback pattern // `asap_types::pre_asap::cse::intern_child` itself uses to recover an @@ -1848,12 +1856,25 @@ pub fn search_workload_with<'s, Id>( (id, expr) }) .collect(); - let cse_roots = share_common_subtrees(owned_roots); + share_common_subtrees(owned_roots) +} +fn search_cse_workload_with<'s, Id>( + cse_roots: Vec<(Id, Rc)>, + strategies: &[Box], +) -> PlanSpace { let mut order = Vec::new(); let mut nodes = HashMap::new(); let mut counts: HashMap<*const QueryExpr, usize> = HashMap::new(); discover_targets(&cse_roots, &mut order, &mut nodes, &mut counts); + let siblings: Vec> = order + .iter() + .filter_map(|ptr| { + let node = &nodes[ptr]; + matches!(node.as_ref(), QueryExpr::Aggregate { .. }).then(|| Rc::clone(node)) + }) + .collect(); + let rollup_strategy = RollupStrategy::new(&siblings); let mut groups: HashMap<*const QueryExpr, MemoGroup> = HashMap::new(); for ptr in &order { @@ -1894,6 +1915,9 @@ pub fn search_workload_with<'s, Id>( proposed.extend(strategy.replacements(&target)); } } + if rollup_strategy.matches(&target) { + proposed.extend(rollup_strategy.replacements(&target)); + } for candidate in &proposed { if let Replacement::Rewrite(rc) = &candidate.replacement { diff --git a/crates/asap-aware-mapping/src/rollup.rs b/crates/asap-aware-mapping/src/rollup.rs index 8f655967..354724cc 100644 --- a/crates/asap-aware-mapping/src/rollup.rs +++ b/crates/asap-aware-mapping/src/rollup.rs @@ -36,14 +36,17 @@ //! - `Sum`/`Min`/`Max` are **self-combining**: a sum of sums is a sum, a min //! of mins is a min. Reapplying the identical `AggIntent` over the finer //! side's own output column is correct. -//! - `Count` is **not** self-combining: re-`Count`ing the finer side's own +//! - Exact `Count` is **not** self-combining: re-`Count`ing the finer side's own //! output rows counts the number of *finer groups* per coarser group (how //! many distinct `region`s a `job` has), not the original row count. The //! correct combinator is `Sum` — `count(A ∪ B) = count(A) + count(B)`, the //! same "`COUNT(*)` over a pre-aggregated summary table becomes //! `SUM(count)`" rule every OLAP rollup/cube/materialized-view-matching -//! implementation applies. `Increase` mirrors it (a counter's total -//! increase across sub-windows is additive), so it combines via `Sum` too. +//! implementation applies. Approximate `Count` is excluded because summing +//! finalized per-finer-group estimates does not preserve the coarser error +//! target. `Increase` mirrors exact count's additive combination (a +//! counter's total increase across sub-windows is additive), so it combines +//! via `Sum` too. //! - `Rate` (`increase / duration`) has **no** valid self- or sum-combinator //! here, so it is deliberately left unhandled (see [`rollup_combinator`]). //! In practice this never matters: `Rate`/`Increase` are constructed with @@ -61,37 +64,27 @@ //! asks for) consults it, and so does this module's `replacements`, so the //! two can never disagree about which intents are eligible. //! -//! ## `ColumnId` comparability — only sound because the child `Rc` is shared +//! ## `ColumnId` comparability — only sound for identical child IR //! //! A `ColumnId` is a *position* into a specific `Schema` (`crates/types/src/pre_asap/schema.rs`'s //! own doc: "the same edge, the same schema, the same positional numbering"). //! Comparing the coarser aggregate's `by` positions against the finer -//! aggregate's `by` positions is only meaningful because **both aggregates -//! share the identical child `Rc`** (post-CSE, per -//! `pre_asap::cse::share_common_subtrees`'s doc on why hash-consing is a -//! precondition, not a coincidence) — so both `by` lists are positional -//! offsets into the exact same schema. Two *structurally different but -//! equivalent* sources (e.g. two independently-built scans of the same -//! underlying table with columns in a different order) are explicitly out -//! of scope: this module never attempts to reconcile `ColumnId`s across two -//! distinct schemas. +//! aggregate's `by` positions is only meaningful when both aggregates have +//! pointer-identical or `PartialEq`-equal children, including equal schemas. +//! Thus both `by` lists index the same shape. The equality fallback matters +//! for scans that CSE conservatively declines to alias because they have no +//! declared unique key. Structurally different sources remain out of scope: +//! this module never reconciles `ColumnId`s across distinct schemas. //! //! ## Non-goals (tracked separately, not attempted here — same split //! `replacement.rs`'s own module docs draw for `SharedSubtreeStrategy`'s //! `consumer_count`) //! -//! - **No workload-wide sibling discovery.** Finding "every `Aggregate` node -//! across a whole workload that shares a given `Rc` child" is a traversal -//! over the *whole* workload, not a fact available from one -//! [`TargetSubDAG`] in isolation — [`ReplacementStrategy::matches`]/ -//! `replacements` only ever see one target at a time. [`RollupStrategy`] -//! takes the already-discovered sibling set as constructor state instead -//! of rediscovering it: a future caller (the search engine, issue #252, -//! built in parallel and possibly not yet landed on this branch) is -//! expected to construct this strategy with the real sibling set it -//! already found — the same "this module wraps a decision, it does not -//! own the traversal that feeds it" split `replacement.rs` draws for -//! `SharedSubtreeStrategy::matches`'s `consumer_count`. +//! - **No sibling discovery inside this strategy.** Finding every aggregate +//! across a workload is not a per-target operation. The workload search +//! performs that traversal after CSE, then constructs [`RollupStrategy`] +//! with the discovered aggregate set. Direct/custom callers supply their +//! own set through [`RollupStrategy::new`]. //! - **No materialized roll-up operator.** Actually building a pre-aggregated //! summary/scan leaf at execution time is separate, larger work outside //! `asap-aware-mapping`'s scope (see issue #254's own "Non-goal" section) @@ -110,6 +103,7 @@ use std::rc::Rc; use asap_types::pre_asap::agg_intent::AggIntent; use asap_types::pre_asap::query_expr::{GroupKeys, QueryExpr, Reduction}; use asap_types::pre_asap::schema::{ColumnId, Schema}; +use asap_types::types::AccuracyTarget; use crate::replacement::{Replacement, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG}; @@ -170,7 +164,10 @@ fn rollup_combinator(intent: &AggIntent, finer_measure_col: ColumnId) -> Option< }), // Not self-combining — see the module docs. Both merge by addition // over the finer side's own output column instead. - AggIntent::Count { .. } | AggIntent::Increase => Some(AggIntent::Sum { + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + } + | AggIntent::Increase => Some(AggIntent::Sum { col: Some(finer_measure_col), }), _ => None, @@ -243,40 +240,43 @@ pub fn is_legal_rollup_source( /// superset, so the length check alone rules out equality without a set /// comparison). fn is_strict_column_superset(finer: &[ColumnId], coarser: &[ColumnId]) -> bool { - if finer.len() <= coarser.len() { + let finer_set: HashSet<&ColumnId> = finer.iter().collect(); + let coarser_set: HashSet<&ColumnId> = coarser.iter().collect(); + if finer_set.len() <= coarser_set.len() { return false; } - let finer_set: HashSet<&ColumnId> = finer.iter().collect(); - coarser.iter().all(|id| finer_set.contains(id)) + coarser_set.is_subset(&finer_set) } /// Wraps the group-by-lattice roll-up reuse (issue #254, part of #33) as a /// [`ReplacementStrategy`]: given a coarser `Aggregate` [`TargetSubDAG`], -/// finds every already-known sibling `Aggregate` that shares the same child -/// `Rc` and is a legal, strictly finer roll-up source for it (per +/// finds every already-known sibling `Aggregate` that has an identical child +/// IR and is a legal, strictly finer roll-up source for it (per /// [`is_legal_rollup_source`]), and proposes replacing the target with a /// re-aggregation over that sibling instead of the shared raw source. /// /// `siblings` is **caller-supplied, not discovered here** — see the module /// docs' "Non-goals" on why finding the full sibling set across a workload /// is a workload-wide traversal this strategy does not own. -pub struct RollupStrategy<'a> { - siblings: &'a [Rc], +pub struct RollupStrategy { + siblings: Vec>, } -impl<'a> RollupStrategy<'a> { - /// A strategy that considers every node in `siblings` as a candidate - /// roll-up source (or target) — typically the full set of `Aggregate` +impl RollupStrategy { + /// A strategy that owns clones of every node in `siblings` and considers + /// each as a candidate roll-up source (or target) — typically the full set of `Aggregate` /// nodes a workload-wide discovery pass (issue #252) already found /// sharing at least one child `Rc` with something else. - pub fn new(siblings: &'a [Rc]) -> Self { - Self { siblings } + pub fn new(siblings: &[Rc]) -> Self { + Self { + siblings: siblings.to_vec(), + } } /// Every sibling that is a legal, strictly finer roll-up source for /// `target` — shared between `matches` and `replacements` so the two /// can never disagree about which siblings qualify. - fn finer_sources(&self, target: &TargetSubDAG<'_>) -> Vec<&'a Rc> { + fn finer_sources(&self, target: &TargetSubDAG<'_>) -> Vec<&Rc> { let Some((coarser_by, coarser_intent, coarser_child)) = bindable_grouped_aggregate(target.root) else { @@ -294,7 +294,7 @@ impl<'a> RollupStrategy<'a> { else { return false; }; - if !Rc::ptr_eq(finer_child, coarser_child) { + if !Rc::ptr_eq(finer_child, coarser_child) && finer_child != coarser_child { return false; } let Ok(finer_schema) = candidate.output_schema() else { @@ -312,7 +312,7 @@ impl<'a> RollupStrategy<'a> { } } -impl ReplacementStrategy for RollupStrategy<'_> { +impl ReplacementStrategy for RollupStrategy { fn matches(&self, target: &TargetSubDAG<'_>) -> bool { !self.finer_sources(target).is_empty() } @@ -513,6 +513,18 @@ mod tests { )); } + #[test] + fn predicate_does_not_treat_duplicate_keys_as_a_strict_superset() { + let finer_schema = Schema::with_time_index(vec![], 0, vec![vec![0, 1]]); + assert!(!is_legal_rollup_source( + &GroupKeys::by(vec![2, 3, 3]), + &finer_schema, + &AggIntent::Sum { col: None }, + &GroupKeys::by(vec![2, 3]), + &AggIntent::Sum { col: None }, + )); + } + #[test] fn predicate_rejects_unrelated_by_sets() { let finer_schema = Schema::with_time_index(vec![], 0, vec![vec![0, 1]]); @@ -634,6 +646,90 @@ mod tests { ); } + #[test] + fn approximate_count_does_not_roll_up_via_sum() { + let scan = Rc::new(metric_scan()); + let intent = AggIntent::Count { + accuracy: AccuracyTarget::Epsilon(0.01), + }; + let fine = agg(vec![2, 3], intent.clone(), &scan); + let coarse = agg(vec![2], intent, &scan); + let siblings = vec![Rc::clone(&fine), Rc::clone(&coarse)]; + let strategy = RollupStrategy::new(&siblings); + let target = TargetSubDAG::new(&coarse); + + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } + + #[test] + fn default_workload_search_adds_rollup_for_two_query_workload() { + let fine_scan = Rc::new(metric_scan()); + let coarse_scan = Rc::new(metric_scan()); + let fine = agg(vec![2, 3], AggIntent::Sum { col: Some(1) }, &fine_scan); + let coarse = agg(vec![2], AggIntent::Sum { col: Some(1) }, &coarse_scan); + + let space = crate::replacement::search_workload(vec![("fine", fine), ("coarse", coarse)]); + let coarse_group = space + .groups() + .find(|group| { + matches!( + group.target.as_ref(), + QueryExpr::Aggregate { + reduction: Reduction::Reduce(by), + .. + } if by.keys() == [2] + ) + }) + .expect("coarser aggregate group"); + + let rewrite = coarse_group + .candidates + .iter() + .find_map(|candidate| match &candidate.replacement { + Replacement::Rewrite(rewrite) => Some(rewrite), + Replacement::Summary(_) => None, + }) + .expect("default search must include the roll-up rewrite"); + let QueryExpr::Aggregate { child, .. } = rewrite.as_ref() else { + panic!("expected aggregate rewrite, got {rewrite:?}"); + }; + assert!(matches!( + child.as_ref(), + QueryExpr::Aggregate { + reduction: Reduction::Reduce(by), + .. + } if by.keys() == [2, 3] + )); + } + + #[test] + fn workload_search_does_not_roll_up_approximate_count() { + let intent = AggIntent::Count { + accuracy: AccuracyTarget::Epsilon(0.01), + }; + let fine = agg(vec![2, 3], intent.clone(), &Rc::new(metric_scan())); + let coarse = agg(vec![2], intent, &Rc::new(metric_scan())); + let space = crate::replacement::search_workload(vec![("fine", fine), ("coarse", coarse)]); + let coarse_group = space + .groups() + .find(|group| { + matches!( + group.target.as_ref(), + QueryExpr::Aggregate { + reduction: Reduction::Reduce(by), + .. + } if by.keys() == [2] + ) + }) + .expect("coarser aggregate group"); + + assert!(coarse_group + .candidates + .iter() + .all(|candidate| !matches!(candidate.replacement, Replacement::Rewrite(_)))); + } + #[test] fn rollup_preserves_the_coarser_output_name() { let scan = Rc::new(metric_scan()); @@ -735,10 +831,10 @@ mod tests { } #[test] - fn different_child_does_not_roll_up_even_with_identical_shape() { - // Two separately-allocated (not CSE-shared) scans: same shape, but - // not the same `Rc`, so their `ColumnId`s are not comparable per - // this module's own scope (see the module docs). + fn structurally_identical_children_roll_up_without_cse_aliasing() { + // Scans without unique keys are deliberately not pointer-aliased by + // CSE. Structural equality still proves identical schemas and makes + // the two aggregates' positional ColumnIds comparable. let fine = agg( vec![2, 3], AggIntent::Sum { col: Some(1) }, @@ -753,7 +849,8 @@ mod tests { let siblings = vec![Rc::clone(&fine), Rc::clone(&coarse)]; let strategy = RollupStrategy::new(&siblings); let target = TargetSubDAG::new(&coarse); - assert!(!strategy.matches(&target)); + assert!(strategy.matches(&target)); + assert_eq!(strategy.replacements(&target).len(), 1); } #[test] From 19718c01fc4ff9de0573643226eba406f8b2fe72 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 25 Aug 2026 09:25:26 -0600 Subject: [PATCH 5/5] docs(mapping): refresh rollup integration status --- crates/asap-aware-mapping/src/explanation.rs | 2 +- crates/asap-aware-mapping/src/lib.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/asap-aware-mapping/src/explanation.rs b/crates/asap-aware-mapping/src/explanation.rs index fa25dd3d..1596634c 100644 --- a/crates/asap-aware-mapping/src/explanation.rs +++ b/crates/asap-aware-mapping/src/explanation.rs @@ -162,7 +162,7 @@ //! | Catalog entry | Status | Where a future `ExplanationKind` would come from | //! |---|---|---| //! | Semantic-equivalent rewriting (e.g. `avg` → `sum`/`count`) | No `ReplacementStrategy` implementation in this codebase yet (tracked separately, issue #253) | Once wired into `default_strategies()`: any `Replacement::Rewrite` candidate that strategy proposes | -//! | Roll-ups (fine-to-coarse group-by reuse) | No `ReplacementStrategy` implementation in this codebase yet (tracked separately, issue #254) | Once wired into `default_strategies()`: any `Replacement::Rewrite` candidate that strategy proposes | +//! | Roll-ups (fine-to-coarse group-by reuse) | [`RollupStrategy`](crate::rollup::RollupStrategy), derived from workload siblings after CSE/target discovery (issue #254) | Any `Replacement::Rewrite` candidate that rolls a coarse aggregate up from a compatible finer aggregate | //! | Wavelets/OMP | Params type exists (`WaveletKind`/`WaveletParams`), reachable only via a deployment `CostModel::realize_extension` (no core `AggIntent` dispatch picks it) | A `ReplacementStrategy` that inspects a deployment's own `CostModel`, once some intent shape actually maps to `Implementation::Wavelet` | //! | Sampling | Same story as Wavelets: `SamplingKind`/`SamplingParams` exist, unreachable from core dispatch | Same hook as Wavelets, for `Implementation::Sample` | //! | Deep generative compression | No representation at all — no `Implementation`/`SummaryFamilyType` variant | Needs a new summary family added to `asap_types::post_asap` first | diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index d7c366db..89fc2ab5 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -101,7 +101,7 @@ //! reuse (issue #254, part of #33) as a [`ReplacementStrategy`]: given a //! coarser `Aggregate` target and a caller-supplied sibling set, proposes //! re-deriving it from an already-computed, strictly finer sibling -//! `Aggregate` over the same shared child instead of an independent pass +//! `Aggregate` over identical child IR instead of an independent pass //! over the raw source — the cross-aggregate sibling of //! `pre_asap::cse::share_common_subtrees`'s identical-subtree sharing. //! [`rollup::is_legal_rollup_source`] is the standalone legality predicate