diff --git a/crates/asap-aware-mapping/src/bind.rs b/crates/asap-aware-mapping/src/bind.rs index e95dd05a..53d605e2 100644 --- a/crates/asap-aware-mapping/src/bind.rs +++ b/crates/asap-aware-mapping/src/bind.rs @@ -45,7 +45,7 @@ use asap_types::pre_asap::schema::Schema; use thiserror::Error; use crate::boundary::{implementation_for_with, Implementation}; -use crate::cost_model::{CostModel, DefaultCostModel}; +use crate::cost_model::{CostModel, CseCandidate, DefaultCostModel, ShareDecision}; /// Errors from the pre-ASAP → post-ASAP binding pass. #[derive(Debug, Error)] @@ -93,8 +93,9 @@ pub fn implement_tree_with( /// Bind a whole workload's worth of already-CSE'd roots /// ([`asap_types::pre_asap::cse::share_common_subtrees`]'s output), reusing -/// one bound [`SummaryNode`] wherever two roots share the same `Rc` (issue -/// #212, #222, #223 stage 2). +/// one bound [`SummaryNode`] wherever two roots share the same `Rc` *and* +/// `cost_model` decides it's worth it (issue #212, #222, #223 stages 2 and +/// 4, #237). /// /// This is a real caller for `share_common_subtrees`, wired up deliberately: /// the pass's own landing plan calls out that its predecessor @@ -111,35 +112,69 @@ pub fn implement_tree_with( /// `asap_aware_mapping::boundary::Matcher`'s documented, deliberately-unfilled /// job, not this one's. /// +/// A first pass over `roots` counts each distinct `Rc` pointer's +/// true `consumer_count` across the whole workload, so the +/// [`CseCandidate`]/[`CostModel::cse_share_decision`] cost comparison (see +/// `docs/cse-cost-model-decision.md`) sees the real total, not a running +/// count that grows as roots are processed left to right. The decision is +/// made once, the first time a shared pointer is bound, and cached alongside +/// the bound `SummaryNode` so every later occurrence of that same `Rc` +/// applies the same decision consistently — either every consumer reuses one +/// shared `SummaryNode`, or every consumer (including the first) binds +/// independently. +/// /// Only whole-root sharing is memoized (matching two workload roots that are /// themselves the same `Rc` after CSE) — [`implement_tree`] is -/// called at most once per distinct root pointer, but it still walks each -/// such tree's own internal structure fresh; a subtree shared only *below* -/// two different roots' top level does not additionally memoize inside that -/// walk. Widening this to sub-root memoization is future work. +/// called at most once per distinct root pointer when the decision is +/// `Share`, but it still walks each such tree's own internal structure +/// fresh; a subtree shared only *below* two different roots' top level does +/// not additionally memoize inside that walk. Widening this to sub-root +/// memoization is future work. pub fn implement_workload( roots: Vec<(Id, Rc)>, ) -> Vec<(Id, Result, ImplementError>)> { implement_workload_with(roots, &DefaultCostModel) } -/// Like [`implement_workload`], but ranks candidate summaries via -/// `cost_model` instead of the built-in static preference order (see +/// Like [`implement_workload`], but ranks candidate summaries — and decides +/// CSE sharing — via `cost_model` instead of the built-in defaults (see /// [`crate::cost_model`]). pub fn implement_workload_with( roots: Vec<(Id, Rc)>, cost_model: &dyn CostModel, ) -> Vec<(Id, Result, ImplementError>)> { - let mut memo: std::collections::HashMap<*const QueryExpr, Rc> = + let mut consumer_count: std::collections::HashMap<*const QueryExpr, usize> = + std::collections::HashMap::new(); + for (_, expr) in &roots { + *consumer_count.entry(Rc::as_ptr(expr)).or_insert(0) += 1; + } + + let mut memo: std::collections::HashMap<*const QueryExpr, (Rc, ShareDecision)> = std::collections::HashMap::new(); roots .into_iter() .map(|(id, expr)| { let ptr = Rc::as_ptr(&expr); let result = match memo.get(&ptr) { - Some(cached) => Ok(Rc::clone(cached)), + Some((cached, ShareDecision::Share)) => Ok(Rc::clone(cached)), + Some((_, ShareDecision::RecomputeIndependently)) => { + implement_tree_with(&expr, cost_model) + } None => implement_tree_with(&expr, cost_model).inspect(|node| { - memo.insert(ptr, Rc::clone(node)); + let count = consumer_count[&ptr]; + let decision = if count > 1 { + let candidate = CseCandidate { + subtree: &expr, + bound_summary: node, + consumer_count: count, + }; + cost_model.cse_share_decision(&candidate) + } else { + // Only one consumer: nothing to compare against, and + // this branch is never consulted again for `ptr`. + ShareDecision::Share + }; + memo.insert(ptr, (Rc::clone(node), decision)); }), }; (id, result) @@ -866,4 +901,43 @@ mod tests { }; assert_eq!(col, &ColumnRef::Named("bytes".into())); } + + /// Issue #237, #223 stage 4: a `CostModel` that declines CSE sharing + /// makes `implement_workload_with` bind each occurrence independently, + /// even though the two roots are the exact same `Rc` (as + /// `share_common_subtrees` would hand back for two identical workload + /// entries) — the opposite of `DefaultCostModel`'s unconditional-share + /// behavior pinned by `crates/integration-tests/tests/cse.rs`. + #[test] + fn implement_workload_with_recomputes_independently_when_cost_model_declines_sharing() { + struct NeverShareCse; + impl CostModel for NeverShareCse { + fn rank_candidates( + &self, + _intent: &AggIntent, + candidates: &[SketchKind], + ) -> Vec { + candidates.to_vec() + } + fn cse_share_decision(&self, _candidate: &CseCandidate) -> ShareDecision { + ShareDecision::RecomputeIndependently + } + } + + let shared = Rc::new(agg(vec![2], default_quantile(0.99), metric_scan(&["job"]))); + let bound = implement_workload_with( + vec![("a", Rc::clone(&shared)), ("b", Rc::clone(&shared))], + &NeverShareCse, + ); + let [(_, ra), (_, rb)] = bound.as_slice() else { + panic!("expected 2 bound results"); + }; + let ra = ra.as_ref().expect("a failed to bind"); + let rb = rb.as_ref().expect("b failed to bind"); + assert!( + !Rc::ptr_eq(ra, rb), + "a CostModel that declines CSE sharing must bind each occurrence \ + independently, even for two roots that are the same Rc" + ); + } } diff --git a/crates/asap-aware-mapping/src/cost_model.rs b/crates/asap-aware-mapping/src/cost_model.rs index 713c79ba..c4944f96 100644 --- a/crates/asap-aware-mapping/src/cost_model.rs +++ b/crates/asap-aware-mapping/src/cost_model.rs @@ -31,13 +31,104 @@ //! [`DefaultCostModel`], so a deployment that never plugs in its own cost //! model keeps today's static-preference-order behavior exactly, byte for //! byte. +//! +//! ## CSE sharing (issue #237, #223 stage 4) +//! +//! [`CseCandidate`]/[`ShareDecision`]/[`CostModel::cse_share_decision`] below +//! decide whether a CSE-detected shared subtree +//! ([`asap_types::pre_asap::cse::share_common_subtrees`], issue #223 stages +//! 1-2, PR #235) is actually worth sharing, via a real Volcano/Cascades-style +//! cost comparison rather than a fixed rule. See +//! `docs/cse-cost-model-decision.md` for the full design discussion (why +//! cost-based, why not a full plan-search engine, the layering constraint +//! that forces detection to stay cost-agnostic). [`bind::implement_workload_with`](crate::bind::implement_workload_with) +//! is the caller. -use asap_types::post_asap::{SketchKind, SketchParams, SketchQuery}; +use asap_types::post_asap::{ + SketchKind, SketchParams, SketchQuery, SummaryFamilyType, SummaryNode, +}; use asap_types::pre_asap::agg_intent::AggIntent; use asap_types::pre_asap::expr_ir::ColumnRef; +use asap_types::pre_asap::query_expr::QueryExpr; use crate::boundary::Implementation; +/// A CSE-detected, legality-gated shared subtree with two or more consumers +/// — the unit [`CostModel::cse_share_decision`] decides over. Built by +/// [`bind::implement_workload_with`](crate::bind::implement_workload_with) +/// the first time it binds a subtree that +/// [`asap_types::pre_asap::cse::share_common_subtrees`] already collapsed +/// onto one `Rc` for two or more workload roots. See +/// `docs/cse-cost-model-decision.md`. +pub struct CseCandidate<'a> { + /// The shared pre-ASAP subtree itself. + pub subtree: &'a QueryExpr, + /// The `SummaryNode` this subtree bound to — gives the cost model the + /// concrete `SummaryFamilyType`/`(kind, params)` actually at stake, not + /// just the pre-ASAP shape. + pub bound_summary: &'a SummaryNode, + /// How many workload roots reference this exact shared subtree, counted + /// once up front over the whole workload (always >= 2 — a candidate is + /// only ever constructed for an actually-shared subtree). + pub consumer_count: usize, +} + +/// The decision [`CostModel::cse_share_decision`] returns for one +/// [`CseCandidate`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ShareDecision { + /// Reuse one bound `SummaryNode` across every consumer. + Share, + /// Bind each occurrence independently — the shared-maintenance cost + /// isn't worth it for this candidate. + RecomputeIndependently, +} + +/// Default [`CostModel::cse_recompute_cost`]: a structural-size proxy — the +/// number of *unique* nodes in `subtree`'s DAG +/// ([`asap_types::pre_asap::cse::dag_node_count`], the same module this +/// candidate's sharing was detected in). Deliberately **not** a raw +/// `serde_json` serialization length: after CSE, `subtree` is generally a +/// DAG, not a tree (a `CseCandidate` only exists because something got +/// shared), and a naive full serialization re-serializes — over-counts — +/// any descendant `subtree` already shares internally, once per parent +/// that references it, instead of once for the whole DAG. `dag_node_count` +/// dedupes by `Rc` pointer identity, so it charges each unique node's +/// contribution exactly once regardless of how many places within +/// `subtree` reference it. Cheap to compute (one pass, no serialization), +/// and still scales with real structural complexity — a genuinely tiny +/// leaf costs little to recompute, a deep multi-join subtree costs a lot. +/// A deployment with real per-row/per-update cost knowledge should +/// override [`CostModel::cse_recompute_cost`] instead of relying on this. +pub fn default_cse_recompute_cost(subtree: &QueryExpr) -> f64 { + asap_types::pre_asap::cse::dag_node_count(subtree) as f64 +} + +/// Default [`CostModel::cse_shared_maintenance_cost`]: a small +/// per-[`SummaryFamilyType`] weight, scaled to the same order of magnitude +/// as [`default_cse_recompute_cost`]'s typical output (a small node +/// count, not a byte length), reflecting that families differ in how +/// expensive they are to keep *continuously updated* for the life of a +/// workload — an exact accumulator is the cheapest (an O(1) merge), +/// sketches/samples cost more (a whole data structure to update per new +/// row), wavelets/fitted models cost the most (coefficient/parameter +/// maintenance). These weights are illustrative, not measured — a +/// deployment with real memory/update-cost numbers should override +/// [`CostModel::cse_shared_maintenance_cost`] instead of relying on this +/// table. +pub fn default_cse_shared_maintenance_cost(family: &SummaryFamilyType) -> f64 { + const UNIT: f64 = 1.0; + let weight = match family { + SummaryFamilyType::Plain(_) => 1.0, + SummaryFamilyType::ExactAggregate(..) => 1.0, + SummaryFamilyType::Sketch(..) => 3.0, + SummaryFamilyType::Sample(..) => 3.0, + SummaryFamilyType::Wavelet(..) => 5.0, + SummaryFamilyType::StatModel(..) => 6.0, + }; + weight * UNIT +} + /// Ranks the candidate summary families for one [`AggIntent`], best choice /// first. /// @@ -116,6 +207,59 @@ pub trait CostModel { readout_extension wasn't overridden to match" ) } + + /// Estimate the one-time cost of recomputing `candidate.subtree` + /// independently at a single use site. Default: + /// [`default_cse_recompute_cost`] (a structural-size proxy). See + /// `docs/cse-cost-model-decision.md`. + fn cse_recompute_cost(&self, candidate: &CseCandidate) -> f64 { + default_cse_recompute_cost(candidate.subtree) + } + + /// Estimate the cost of maintaining `candidate.bound_summary` as one + /// continuously-updated shared summary for the life of the workload. + /// Default: [`default_cse_shared_maintenance_cost`] (a per-family + /// weight table), applied to whichever field of + /// `candidate.bound_summary`'s output schema actually carries summary + /// state (falls back to the cheapest, `Plain`, weight if none does — + /// e.g. `bound_summary` is a passthrough `Logical` node with nothing + /// summary-shaped to maintain). See `docs/cse-cost-model-decision.md`. + fn cse_shared_maintenance_cost(&self, candidate: &CseCandidate) -> f64 { + let family = candidate + .bound_summary + .schema + .fields + .iter() + .map(|f| &f.dtype) + .find(|dtype| !matches!(dtype, SummaryFamilyType::Plain(_))) + .cloned() + .unwrap_or(SummaryFamilyType::Plain( + asap_types::pre_asap::DataType::Float64, + )); + default_cse_shared_maintenance_cost(&family) + } + + /// Decide whether to reuse one shared `SummaryNode` across every + /// consumer of `candidate`, or bind each occurrence independently — a + /// Volcano/Cascades-style cost comparison (issue #237, #223 stage 4; see + /// `docs/cse-cost-model-decision.md`): share iff the estimated cost of + /// maintaining one shared summary is no greater than the estimated total + /// cost of recomputing it independently everywhere it's used. + /// + /// The default body composes [`cse_recompute_cost`](Self::cse_recompute_cost) + /// and [`cse_shared_maintenance_cost`](Self::cse_shared_maintenance_cost) + /// — a deployment with real cost knowledge should override those two + /// (keeping this comparison), or override this method directly for a + /// wholly different policy. + fn cse_share_decision(&self, candidate: &CseCandidate) -> ShareDecision { + let recompute_total = self.cse_recompute_cost(candidate) * candidate.consumer_count as f64; + let shared = self.cse_shared_maintenance_cost(candidate); + if shared <= recompute_total { + ShareDecision::Share + } else { + ShareDecision::RecomputeIndependently + } + } } /// The default cost model: preserves [`summary_candidates`]'s built-in static @@ -227,4 +371,197 @@ mod tests { crate::boundary::default_size_params(SketchKind::Hll, &intent, 0.01, 0.01), ); } + + // ── CSE sharing (issue #237, #223 stage 4) ────────────────────────── + + use asap_types::post_asap::{ExactKind, ExactParams, SummaryExpr, SummaryField, SummarySchema}; + use asap_types::pre_asap::query_expr::Source; + use asap_types::pre_asap::schema::{Column, DataType, Schema}; + + fn 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), + ], + 0, + vec![], + ), + } + } + + fn summary_node(family: SummaryFamilyType) -> SummaryNode { + SummaryNode { + expr: SummaryExpr::SummaryAgg { + child: std::rc::Rc::new(SummaryNode { + expr: SummaryExpr::Logical(Box::new(scan())), + schema: SummarySchema { + fields: vec![], + time_index: None, + }, + }), + family: family.clone(), + col: asap_types::pre_asap::expr_ir::ColumnRef::Named("value".into()), + reduction: asap_types::pre_asap::query_expr::Reduction::by(vec![]), + }, + schema: SummarySchema { + fields: vec![SummaryField { + name: "state".into(), + dtype: family, + nullable: false, + }], + time_index: None, + }, + } + } + + #[test] + fn default_recompute_cost_is_positive_and_grows_with_structural_size() { + let leaf = scan(); + let nested = QueryExpr::Dedup { + cols: vec![0], + child: std::rc::Rc::new(leaf.clone()), + }; + assert!(default_cse_recompute_cost(&leaf) > 0.0); + assert!(default_cse_recompute_cost(&nested) > default_cse_recompute_cost(&leaf)); + } + + /// The DAG-awareness this proxy exists for: a subtree that internally + /// re-references one shared descendant (e.g. after single-query CSE, + /// `x op x` collapsing both branches onto one `Rc`) must cost the same + /// as if that descendant only appeared once — not double, the way a + /// naive tree-shaped size measure (a full serialization, or an + /// identity-blind recursive walk) would count it. + #[test] + fn default_recompute_cost_does_not_double_count_an_internally_shared_descendant() { + use asap_types::pre_asap::expr_ir::ScalarValue; + use asap_types::pre_asap::query_expr::{JoinKind, Predicate}; + + let true_pred = || { + Predicate(std::rc::Rc::new(QueryExpr::Literal(ScalarValue::Boolean( + true, + )))) + }; + let shared_leaf = std::rc::Rc::new(scan()); + let no_sharing = QueryExpr::Join { + kind: JoinKind::Inner, + pred: true_pred(), + left: std::rc::Rc::new(scan()), + right: std::rc::Rc::new(scan()), + }; + let with_sharing = QueryExpr::Join { + kind: JoinKind::Inner, + pred: true_pred(), + left: std::rc::Rc::clone(&shared_leaf), + right: std::rc::Rc::clone(&shared_leaf), + }; + assert_eq!( + default_cse_recompute_cost(&no_sharing), + 3.0, + "no sharing: Join + 2 independent Scans = 3 unique nodes" + ); + assert_eq!( + default_cse_recompute_cost(&with_sharing), + 2.0, + "internal sharing: Join + 1 shared Scan (referenced twice) = \ + 2 unique nodes, not 3 — a tree-shaped size measure would \ + wrongly charge for the shared Scan twice" + ); + } + + #[test] + fn default_shared_maintenance_cost_orders_families_cheapest_to_priciest() { + let exact = default_cse_shared_maintenance_cost(&SummaryFamilyType::ExactAggregate( + ExactKind::Sum, + ExactParams::Sum, + )); + let sketch = default_cse_shared_maintenance_cost(&SummaryFamilyType::Sketch( + SketchKind::Hll, + SketchParams::Hll { precision: 12 }, + )); + assert!( + exact < sketch, + "an exact accumulator should be cheaper to keep continuously updated \ + than a sketch: exact={exact}, sketch={sketch}" + ); + } + + #[test] + fn cse_share_decision_shares_when_recompute_dominates_maintenance() { + let candidate = CseCandidate { + subtree: &scan(), + bound_summary: &summary_node(SummaryFamilyType::ExactAggregate( + ExactKind::Sum, + ExactParams::Sum, + )), + // Many consumers of a cheap accumulator: recompute_total should + // dominate the fixed maintenance cost. + consumer_count: 1000, + }; + assert_eq!( + DefaultCostModel.cse_share_decision(&candidate), + ShareDecision::Share + ); + } + + #[test] + fn cse_share_decision_recomputes_when_maintenance_dominates_recompute() { + let candidate = CseCandidate { + subtree: &scan(), + bound_summary: &summary_node(SummaryFamilyType::StatModel( + asap_types::post_asap::StatModelKind::Parametric, + asap_types::post_asap::StatModelParams::Parametric { + family: "gaussian_mixture".into(), + }, + )), + // A single, cheap-to-recompute leaf (scan() alone is 1 DAG + // node, recompute_total = 1) against an expensive-to-maintain + // family (StatModel, maintenance cost 6.0): maintenance should + // dominate. + consumer_count: 1, + }; + assert_eq!( + DefaultCostModel.cse_share_decision(&candidate), + ShareDecision::RecomputeIndependently + ); + } + + #[test] + fn cse_share_decision_default_body_composes_the_two_cost_hooks() { + struct AlwaysExpensiveToRecompute; + impl CostModel for AlwaysExpensiveToRecompute { + fn rank_candidates( + &self, + _intent: &AggIntent, + candidates: &[SketchKind], + ) -> Vec { + candidates.to_vec() + } + fn cse_recompute_cost(&self, _candidate: &CseCandidate) -> f64 { + 1e9 + } + } + + // Even the priciest family should lose to an overridden recompute + // cost this large, confirming `cse_share_decision`'s default body + // actually calls through to the overridable hooks rather than + // hardcoding a comparison against its own defaults. + let candidate = CseCandidate { + subtree: &scan(), + bound_summary: &summary_node(SummaryFamilyType::StatModel( + asap_types::post_asap::StatModelKind::Parametric, + asap_types::post_asap::StatModelParams::Parametric { + family: "gaussian_mixture".into(), + }, + )), + consumer_count: 2, + }; + assert_eq!( + AlwaysExpensiveToRecompute.cse_share_decision(&candidate), + ShareDecision::Share + ); + } } diff --git a/crates/types/src/pre_asap/cse.rs b/crates/types/src/pre_asap/cse.rs index 13100850..c807f534 100644 --- a/crates/types/src/pre_asap/cse.rs +++ b/crates/types/src/pre_asap/cse.rs @@ -86,6 +86,15 @@ //! `asap-plan::cse::dedupe_subtrees` was deleted in #192 for exactly that). //! Stages 3 (`dag_export::structural_hash` unification) and 4 (`CostModel` //! CSE credit) are deliberately deferred follow-ups, not attempted here. +//! Stage 4 (issue #237) is implemented in +//! `asap_aware_mapping::cost_model::CostModel::cse_share_decision` and its +//! caller, `asap_aware_mapping::bind::implement_workload_with` — a real, +//! Volcano/Cascades-style cost comparison over what this module detects, not +//! a fixed rule. See `docs/cse-cost-model-decision.md`. This module's own +//! unconditional "share whenever legal" behavior is unchanged: detection +//! stays cost-agnostic by construction (this crate cannot depend on +//! `asap-aware-mapping`'s `CostModel`), and the cost-aware decision is +//! applied downstream, after detection, over what this module finds. use std::collections::HashMap; use std::hash::{Hash, Hasher}; @@ -155,6 +164,113 @@ fn structural_hash(node: &QueryExpr) -> u64 { hasher.finish() } +/// Count of *unique* nodes reachable from `root`, deduplicated by `Rc` +/// pointer identity (`Rc::as_ptr`) — the real size of the DAG rooted at +/// `root`, not a tree-walk count. +/// +/// After [`share_common_subtrees`] runs (or even before it, for a tree a +/// front end already built with internal `Rc` sharing — e.g. re-running +/// CSE, or a single-query repeated subexpression), `root` is generally a +/// **DAG**, not a tree — that is this whole module's premise. Anything that +/// walks `root` as if every reference were a fresh subtree (a naive +/// recursive walk with no identity tracking, or a naive full +/// `serde_json` serialization — `Rc`'s `Serialize` impl serializes the +/// pointee's *value* at every occurrence, it does not dedupe by identity) +/// re-visits/re-counts an already-shared descendant once per parent that +/// references it, over-counting relative to the actual work of holding it +/// in memory or recomputing it once. This function is the DAG-correct +/// alternative: each unique node is counted exactly once, regardless of +/// how many places within `root` reference it. +/// +/// `pub` so cost-aware callers outside this crate (e.g. +/// `asap_aware_mapping::CostModel::cse_recompute_cost`'s default) have a +/// DAG-correct structural-size proxy available, instead of reaching for +/// something tree-shaped like a raw serialization length. +/// +/// Same operator-child traversal scope as [`share_common_subtrees`] itself +/// (see the module doc's "Algorithm" section, and this module's private +/// `rebuild_children`) — a scalar subexpression embedded in a wrapper +/// position (`Predicate`, `ProjectItem.expr`, `Aggregate.having`, …) is not +/// separately visited, matching this module's own stated scope; it's +/// counted as part of its owning operator node, the same node +/// `rebuild_children` treats as a single opaque leaf for interning +/// purposes. +pub fn dag_node_count(root: &QueryExpr) -> usize { + let mut seen: std::collections::HashSet<*const QueryExpr> = std::collections::HashSet::new(); + count_unique(root, &mut seen) +} + +/// One node's own contribution (`1`) plus each *not-yet-seen* operator +/// child's contribution — exhaustive over every `QueryExpr` variant, +/// enumerating the same fields [`rebuild_children`] does (kept as a +/// separate, read-only traversal rather than threaded through +/// `rebuild_children` itself, since that function consumes and rebuilds +/// its input while this one only ever reads it). +fn count_unique(node: &QueryExpr, seen: &mut std::collections::HashSet<*const QueryExpr>) -> usize { + use QueryExpr::*; + + /// Visit one `Rc`-held child: counts (and recurses into) it only the + /// first time its pointer is seen, `0` on every later occurrence — + /// this is the actual dedup step. + fn visit( + child: &Rc, + seen: &mut std::collections::HashSet<*const QueryExpr>, + ) -> usize { + if seen.insert(Rc::as_ptr(child)) { + count_unique(child, seen) + } else { + 0 + } + } + + 1 + match node { + // `PromqlScalarBridge`'s child is a scalar-sub-language node (issue + // #220), never descended into — same treatment `rebuild_children` + // gives it (see that function's comment on this same variant). + Scan { .. } | PromqlScalarBridge(_) | QueryTimestamp => 0, + PromqlVectorFromScalar(c) | PromqlScalarFromVector(c) => visit(c, seen), + PromqlRelabel { child, .. } + | PromqlInfoEnrich { child, .. } + | PromqlSeriesSample { child, .. } + | Filter { child, .. } + | Project { child, .. } + | Aggregate { child, .. } + | Dedup { child, .. } + | Sort { child, .. } + | Limit { child, .. } + | PromqlSubquery { child, .. } + | TimeRange { child, .. } + | TimeShift { child, .. } + | SQLWindowFunc { child, .. } => visit(child, seen), + // `Concat`'s branches are stored by value (`Vec`, not + // `Rc` — see `rebuild_children`'s `intern_owned` use for + // this variant), so a branch has no `Rc` identity of its own to + // dedup on at this position; still recurse into each in case an + // `Rc`-shared descendant appears further down. + Concat { children } => children.iter().map(|c| count_unique(c, seen)).sum(), + Join { left, right, .. } | SetOp { left, right, .. } => { + visit(left, seen) + visit(right, seen) + } + BinaryOp { lhs, rhs, .. } => visit(lhs, seen) + visit(rhs, seen), + // Scalar variants (issue #205) — never descended into, matching + // `rebuild_children`'s own scope exactly (see its trailing match + // arm and this module's "Algorithm" section). + Column(_) + | Literal(_) + | Compare { .. } + | BoolAnd(_) + | BoolOr(_) + | Not(_) + | IsNull(_) + | IsNotNull(_) + | Cast { .. } + | InList { .. } + | FunctionCall { .. } + | Arithmetic { .. } + | Case { .. } => 0, + } +} + /// Recurse into `child`, then intern the result. `Rc::try_unwrap` recovers /// the owned node without cloning in the overwhelmingly common case — a /// tree freshly built by a front end / `resolve_root`, not yet shared by any @@ -507,6 +623,67 @@ mod tests { ); } + // ── dag_node_count ─────────────────────────────────────────────────── + + #[test] + fn dag_node_count_is_the_naive_count_when_nothing_is_shared() { + // scan() alone: 1 node. + assert_eq!(dag_node_count(&scan()), 1); + // quantile_agg's own child is a fresh, unshared scan(): 2 nodes. + assert_eq!(dag_node_count(&quantile_agg(vec![1], Some(2), 0.5)), 2); + } + + #[test] + fn dag_node_count_deduplicates_an_internally_shared_subtree() { + // Same shape as `single_query_shares_its_own_repeated_subtree`: a + // BinaryOp whose two branches are the *same* Rc after + // `share_common_subtrees` (2 nodes: Scan + Aggregate) — the root + // itself makes 3 unique nodes total (BinaryOp, Aggregate, Scan), + // not 5 (which a tree-walk / naive serialization, counting the + // shared branch's 2 nodes twice, would report). + let agg = quantile_agg(vec![1], Some(2), 0.5); + let root = QueryExpr::BinaryOp { + op: BinaryOpKind::Compare(crate::pre_asap::expr_ir::CompareOpKind::Eq), + lhs: Rc::new(agg.clone()), + rhs: Rc::new(agg), + vector_match: None, + }; + let shared = share_common_subtrees(vec![("q", root)]); + let [(_, root)] = shared.as_slice() else { + panic!("expected 1 root"); + }; + assert_eq!( + dag_node_count(root), + 3, + "the shared branch's 2 nodes must be counted once, not once per \ + occurrence — got {} for {root:?}", + dag_node_count(root) + ); + } + + #[test] + fn dag_node_count_deduplicates_across_two_workload_roots() { + // Two workload roots sharing one Aggregate after + // `share_common_subtrees` (the `duplicate_workload_queries_...` + // shape from `crates/integration-tests/tests/cse.rs`, built + // directly here): each root's own `dag_node_count` must report the + // shared subtree's real size once, not double-count anything — + // there's nothing *to* double-count from a single root's own count + // in this case (no root references the shared node twice), so this + // pins the simpler, more common case that a per-candidate cost + // proxy (`CseCandidate::subtree` in `asap-aware-mapping`) actually + // exercises: counting one occurrence's own reachable DAG size. + let a = quantile_agg(vec![1], Some(2), 0.5); + let b = quantile_agg(vec![1], Some(2), 0.5); + let shared = share_common_subtrees(vec![("a", a), ("b", b)]); + let [(_, ra), (_, rb)] = shared.as_slice() else { + panic!("expected 2 roots"); + }; + assert!(Rc::ptr_eq(ra, rb), "fixture sanity: the two roots merged"); + assert_eq!(dag_node_count(ra), 2); + assert_eq!(dag_node_count(rb), 2); + } + #[test] fn dedup_gates_sharing_the_same_as_aggregate() { // `Dedup { cols }` adds `cols` as a unique key — so two identical diff --git a/docs/cse-cost-model-decision.md b/docs/cse-cost-model-decision.md new file mode 100644 index 00000000..edc5731c --- /dev/null +++ b/docs/cse-cost-model-decision.md @@ -0,0 +1,104 @@ +# CSE sharing: rule-based vs. cost-based framework (issue #237) + +## Context + +[`asap_types::pre_asap::cse::share_common_subtrees`](../crates/types/src/pre_asap/cse.rs) +(issue #223 stages 1-2, PR #235) already *detects* every structurally-identical, +legally-shareable (`Schema::unique_keys`-gated) subtree and shares it +**unconditionally** — there is no cost gate on top of legality. This document +decides the framework for stage 4, "wire workload-level CSE credit into +`CostModel`" — turning "these two subtrees are the same computation" into +"and it's actually worth maintaining one shared summary for them." + +## The two textbook framings (as posed in #237) + +| Framework | Mechanism | CSE policy | +|---|---|---| +| Volcano/Cascades (SQL Server, Snowflake, Calcite) | cost-based: explores a plan space via DP + memo | share iff a real cost comparison (materialize/maintain vs. recompute-per-site) favors it | +| System R (classic) | heuristic: fixed rules over basic statistics | share whenever a fixed rule says to (e.g. "referenced more than once"), no per-case comparison | + +## Decision: cost-based (Volcano/Cascades), implemented for real + +This lands as an actual cost comparison, not a documented-but-unimplemented +shape. [`CostModel::cse_share_decision`](../crates/asap-aware-mapping/src/cost_model.rs) +compares two real, overridable cost estimates for every CSE candidate with +two or more consumers: + +- `cse_recompute_cost(candidate) * candidate.consumer_count` — the total cost + of recomputing the subtree independently at every use site. +- `cse_shared_maintenance_cost(candidate)` — the cost of keeping one shared + summary alive and continuously updated for the workload's lifetime. + +Share iff the shared-maintenance cost is no greater than the total recompute +cost. This is a genuine Volcano/Cascades-style decision: a real, per-candidate +cost comparison, not a fixed "always share when legal" rule. + +Why cost-based and not pure System R: a shared summary here is not a free win +the way sharing a relational scan is in a textbook OLTP optimizer — it is a +sketch/accumulator that (per this crate's stated purpose: *workload*-level +planning, not single-query) is typically kept **continuously updated** as new +data arrives, for as long as the workload runs, regardless of how often it's +actually read. A structurally-shareable subtree that is cheap to recompute on +demand, or rarely queried, can cost more to keep alive as a standing shared +summary than to just recompute independently at each of its (few, or cheap) +use sites. A blanket "always share" rule cannot express that trade-off; a +cost comparison does, without needing a separately hardcoded cheap-threshold +carve-out — a cheap-to-recompute candidate naturally loses the comparison on +its own. + +Why this doesn't need full Volcano/Cascades-scale infrastructure: this repo +has no plan-enumeration/DP-search engine anywhere, and `CostModel` is +deliberately a narrow, single-shot ranking/sizing interface +(`rank_candidates`/`size_params`), not a cost-driven search engine. The +decision here is binary (share vs. don't, per already-detected candidate), +so a direct cost comparison captures the Volcano/Cascades *policy* — weigh +real costs, don't apply a fixed rule — without requiring a memo-based search +space this repo doesn't otherwise have. `implement_workload_with` still +computes the true `consumer_count` for each candidate via a whole-workload +pre-pass before deciding, rather than deciding on a running/partial count — +the decision is made once, from full knowledge of the workload's sharing +structure, the same way a real cost-based optimizer would. + +## Layering constraint + +`share_common_subtrees` lives in `asap-types::pre_asap` — a lower layer that +`asap-aware-mapping` (which owns `CostModel`) depends on, never the reverse. +Detection therefore cannot consult cost even if it wanted to. This is why +stage 1/2's detection stays unconditional (correctly, as a legality-only +gate) and the cost-aware decision is applied downstream, in +`asap-aware-mapping`, after detection rather than fused into it. + +## Where it hooks in + +[`bind::implement_workload_with`](../crates/asap-aware-mapping/src/bind.rs) +computes each shared subtree's true `consumer_count` across the whole +workload up front, then — the first time it binds that subtree — asks +`CostModel::cse_share_decision` once and caches the resulting `ShareDecision` +alongside the bound `SummaryNode`, so every later occurrence of the same +`Rc` consistently reuses the cached summary (`Share`) or rebinds +independently (`RecomputeIndependently`) per that one decision. + +## Defaults + +`cse_recompute_cost`'s default is a structural-size proxy: `cse::dag_node_count`, +the number of *unique* nodes in the subtree's DAG (deduplicated by `Rc` +pointer identity), not a raw serialization length. This distinction matters +here specifically — a `CseCandidate`'s subtree is, by definition, something +CSE already found sharing in, so it's generally a DAG, not a tree; a naive +tree-shaped size measure (a full `serde_json` serialization, or a recursive +walk with no identity tracking) would re-count any descendant the subtree +already shares internally once per parent that reaches it, over-stating the +real cost of holding or recomputing it once. `cse_shared_maintenance_cost`'s default +is a small per-`SummaryFamilyType` weight table (exact accumulators cheapest, +sketches/samples/wavelets/stat-models progressively more expensive to keep +continuously updated) scaled to the same order of magnitude as typical +subtree sizes. Both are documented as coarse heuristic proxies — a real +deployment with actual memory/update-cost/query-frequency knowledge overrides +either or both, same as `size_params` already lets a deployment override +`asap-plan`'s built-in sizing formulas without forking anything else. + +## Scope + +This decision, and `cse_share_decision`'s wiring into `implement_workload_with`, +close out #223's stage 4 and #212's original "add CSE" tracking issue. Stage +3 (`dag_export::structural_hash` unification) landed separately in PR #244.