diff --git a/crates/asap-aware-mapping/src/cost_model.rs b/crates/asap-aware-mapping/src/cost_model.rs index c02adf3a..bf1d5998 100644 --- a/crates/asap-aware-mapping/src/cost_model.rs +++ b/crates/asap-aware-mapping/src/cost_model.rs @@ -49,7 +49,8 @@ use std::rc::Rc; use asap_types::post_asap::{ - SketchAlgorithm, SketchParams, SketchQuery, SummaryFamilyType, SummaryNode, + GroupingStrategy, HydraParams, SketchAlgorithm, SketchParams, SketchQuery, SummaryExpr, + SummaryFamilyType, SummaryNode, }; use asap_types::pre_asap::agg_intent::AggIntent; use asap_types::pre_asap::expr_ir::ColumnRef; @@ -222,6 +223,37 @@ pub trait CostModel { crate::replacement::default_size_params(kind, intent, eps, delta) } + /// Estimated number of distinct subpopulations produced by `target`'s + /// grouping keys. `None` means the deployment has no cardinality estimate; + /// grouping alternatives remain legal but keep their discovery order. + fn estimated_subpopulation_count(&self, _target: &QueryExpr) -> Option { + None + } + + /// Comparable memory-state cost for a sketch grouping candidate. The + /// default compares `N` independent inner sketches against the complete + /// shared Hydra grid, using [`Self::estimated_subpopulation_count`]. + fn grouping_state_cost( + &self, + candidate: &ReplacementSubDAG, + target: &TargetSubDAG<'_>, + ) -> Option { + let Replacement::Summary(node) = &candidate.replacement else { + return None; + }; + let (kind, grouping) = sketch_state(node)?; + let inner = sketch_state_units(kind.params()); + let units = match grouping { + GroupingStrategy::PerSubpopulationInstance => { + inner * self.estimated_subpopulation_count(target.root)? as f64 + } + GroupingStrategy::SharedMultiSubpopulation { params, .. } => { + inner * hydra_grid_cells(params) + } + }; + Some(Cost(units)) + } + /// Realize an `AggIntent::Extension { ext_kind, payload }` — a /// deployment-specific intent shape core has no realization opinion /// for (issue #131). `replacement::implementations_for_with` consults this @@ -348,6 +380,47 @@ pub trait CostModel { } } +fn sketch_state( + node: &SummaryNode, +) -> Option<(&asap_types::post_asap::SketchKind, &GroupingStrategy)> { + match &node.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => sketch_state(summary_input), + SummaryExpr::SummaryAgg { + family: SummaryFamilyType::Sketch(kind, grouping), + .. + } => Some((kind, grouping)), + _ => None, + } +} + +fn sketch_state_units(params: &SketchParams) -> f64 { + match params { + SketchParams::Cms { width, depth } + | SketchParams::CountSketch { width, depth } + | SketchParams::CmsWithHeap { width, depth, .. } + | SketchParams::CountSketchWithHeap { width, depth, .. } => { + f64::from(*width) * f64::from(*depth) + } + _ => 1.0, + } +} + +fn hydra_grid_cells(params: &HydraParams) -> f64 { + match params { + HydraParams::HydraKll { shared_buckets, .. } => f64::from(*shared_buckets), + HydraParams::HydraCms { + shared_rows, + shared_columns, + .. + } + | HydraParams::HydraCountSketch { + shared_rows, + shared_columns, + .. + } => f64::from(*shared_rows) * f64::from(*shared_columns), + } +} + /// Apply [`CostModel::rank_candidates`] and enforce its permutation-only /// contract at the boundary where planner code consumes the result. pub(crate) fn validated_candidate_ranking( @@ -578,7 +651,8 @@ mod tests { // ── CSE sharing (issue #237, #223 stage 4) ────────────────────────── use asap_types::post_asap::{ - ExactKind, ExactParams, SketchKind, SummaryExpr, SummaryField, SummarySchema, + ExactKind, ExactParams, GroupingStrategy, SketchKind, SummaryExpr, SummaryField, + SummarySchema, }; use asap_types::pre_asap::query_expr::Source; use asap_types::pre_asap::schema::{Column, DataType, Schema}; @@ -611,6 +685,7 @@ mod tests { family: family.clone(), col: asap_types::pre_asap::expr_ir::ColumnRef::Named("value".into()), reduction: asap_types::pre_asap::query_expr::Reduction::by(vec![]), + grouping: GroupingStrategy::default(), }, schema: SummarySchema { fields: vec![SummaryField { @@ -685,6 +760,7 @@ mod tests { )); let sketch = default_cse_shared_maintenance_cost(&SummaryFamilyType::Sketch( SketchKind::new(SketchAlgorithm::Hll, SketchParams::Hll { precision: 12 }), + GroupingStrategy::default(), )); assert!( exact < sketch, diff --git a/crates/asap-aware-mapping/src/grouping.rs b/crates/asap-aware-mapping/src/grouping.rs new file mode 100644 index 00000000..f4896329 --- /dev/null +++ b/crates/asap-aware-mapping/src/grouping.rs @@ -0,0 +1,609 @@ +//! `GroupingStrategy` (issue #256, part of #33): the axis deciding whether a +//! grouped aggregate's summary state is built as one independent instance +//! per `by` subpopulation (today's only, implicit behavior) or as one +//! shared Hydra-family structure serving all of them — orthogonal to +//! *which* summary family/kind answers the intent, the same way +//! [`asap_types::post_asap::GroupingStrategy`]'s own doc explains. +//! +//! ## Placement: planning metadata and edge-state type +//! +//! `SummaryExpr::SummaryAgg` carries the grouping choice next to the +//! `Reduction` whose `by` keys determine legality. The same choice is also +//! committed to `SummaryFamilyType::Sketch` on the aggregate's output edge. +//! That duplication is intentional: the node field makes the choice easy to +//! inspect during planning, while the edge type ensures an independent KLL/ +//! CMS state and a Hydra-backed state cannot be accepted as compatible inputs +//! to a downstream `SummaryMerge`. [`with_grouping`] updates both atomically. +//! +//! ## Legality vs. cost (same split [`crate::replacement::implementations_for_with`] +//! already draws) +//! +//! This module only answers "is `SharedMultiSubpopulation` valid here at +//! all", never "is it worth it": +//! +//! - **Non-empty `by`** ([`has_subpopulations`]): an aggregate with no +//! subpopulation concept (a global reduction, or a per-entity reduction +//! with no grouping concept at all) has nothing for a +//! shared-multi-subpopulation structure to multiplex across. +//! - **The family has a Hydra variant** +//! ([`asap_types::post_asap::hydra_kind_for`]): only `Cms` and +//! `CountSketch` are selectable today because their error guarantees are +//! modeled. `HydraKll` remains an explicit experimental IR value, but the +//! paper excludes quantiles and search therefore never emits it. +//! +//! Whether Hydra is *worth it* for a given estimated subpopulation +//! cardinality is a cost-model question, deliberately out of scope here — +//! candidates with no modeled error bound are excluded before costing. +//! +//! ## No `ForceSketchKind`-style steering — bind one already-known candidate directly +//! +//! An earlier draft of this module (written against the very first draft of +//! #251) reused a `CostModel`-wrapping adapter that "steered" a +//! whole-recursive-bind decision procedure toward a specific `SketchKind`, +//! the same pattern [`crate::replacement::SketchAlgorithmStrategy`]'s own module +//! docs explain was deliberately deleted from this crate as an anti-pattern: +//! forcing a choice via a whole-tree `CostModel` adapter had a real bug where +//! the forced choice could leak into a target's own nested aggregates. This +//! module never needs that: [`crate::replacement::implementations_for_with`] +//! already returns every ranked candidate `Implementation` directly, so +//! [`build_candidate`](HydraGroupingStrategy::build_candidate) just finds the +//! one whose `Implementation::Sketch(kind)` has `kind.algorithm()` matching +//! the Hydra-eligible `sketch_kind` it's building a candidate for, and +//! passes that exact, +//! already-decided `Implementation` to +//! [`crate::replacement::construct_summary`] — the same first-class, +//! one-candidate-at-a-time primitive [`crate::replacement::SketchAlgorithmStrategy`] +//! itself calls once per candidate. No adapter, no steering, no risk of a +//! forced choice leaking into nested aggregates. +//! +//! ## Cross-axis legality with roll-up (issue #254) +//! +//! Roll-up and Hydra currently operate on disjoint candidates. Roll-up +//! rewrites exact `Sum`/`Min`/`Max`/`Count` aggregates in the pre-ASAP DAG; +//! Hydra is offered only for approximate quantile/count intents (KLL, CMS, +//! Count-Sketch) and produces a terminal post-ASAP summary candidate. +//! Consequently neither strategy can +//! presently offer the other's candidate as a source. If roll-up support is +//! extended to mergeable sketches, that extension must consult +//! `rollup::is_legal_rollup_source` and add explicit Hydra merge semantics; +//! grouping alone must not imply that a sketch can be rolled up. + +use std::rc::Rc; + +use asap_types::post_asap::{ + default_hydra_params, hydra_kind_for, GroupingStrategy, HydraKind, SketchAlgorithm, + SketchParams, SummaryExpr, SummaryFamilyType, SummaryNode, +}; +use asap_types::pre_asap::agg_intent::AggIntent; +use asap_types::pre_asap::query_expr::{QueryExpr, Reduction}; + +use crate::cost_model::{CostModel, DefaultCostModel}; +use crate::replacement::{ + bindable_intent, construct_summary, describe_intent, implementations_for_with, + summary_candidates, Implementation, Replacement, ReplacementStrategy, ReplacementSubDAG, + TargetSubDAG, +}; + +/// Whether `reduction` has a genuine subpopulation concept for +/// `GroupingStrategy::SharedMultiSubpopulation` to multiplex across — the +/// non-empty-`by` legality condition issue #256 requires. +/// +/// - [`Reduction::PerEntity`]: no grouping concept at all (never merges +/// across entities) — `false`. +/// - [`Reduction::Reduce`] with an empty, non-`without` `by`: a genuine full +/// reduction, one output row, no subpopulations — `false`. +/// - [`Reduction::Reduce`] with a non-empty `by`, or any `without(...)` +/// exclusion grouping (which groups by whatever labels remain, even +/// `without([])` — "group by every label"): a real subpopulation concept +/// — `true`. +pub fn has_subpopulations(reduction: &Reduction) -> bool { + match reduction.group_keys() { + None => false, + Some(keys) => keys.is_without() || !keys.is_empty(), + } +} + +/// A single static instance so [`HydraGroupingStrategy::default_cost_model`] +/// can hand out a `&'static dyn CostModel` without heap-allocating one — same +/// pattern [`crate::replacement::SketchAlgorithmStrategy`] uses. +static DEFAULT_COST_MODEL: DefaultCostModel = DefaultCostModel; + +/// Wraps the `GroupingStrategy` axis (issue #256) as a +/// [`ReplacementStrategy`]: for a target [`SketchAlgorithmStrategy`](crate::replacement::SketchAlgorithmStrategy) +/// already has an opinion on, offers an additional +/// `GroupingStrategy::SharedMultiSubpopulation` candidate wherever the +/// legality conditions in the module docs above hold — alongside, not +/// instead of, the per-subpopulation candidates `SketchAlgorithmStrategy` +/// itself enumerates. The workload search composes both strategies over the +/// same target, so it sees every summary-family alternative *and* the Hydra +/// alternative; the built-in workload search registers both strategies, and +/// this strategy's own `replacements()` reports only the +/// latter, matching every other strategy in this crate's "one strategy, one +/// concern" shape. +pub struct HydraGroupingStrategy<'a> { + cost_model: &'a dyn CostModel, +} + +impl HydraGroupingStrategy<'static> { + /// A strategy that ranks/binds via the built-in [`DefaultCostModel`] — + /// what a deployment gets with no custom cost model plugged in, the same + /// default [`crate::replacement::SketchAlgorithmStrategy::default_cost_model`] + /// offers. + pub fn default_cost_model() -> Self { + Self { + cost_model: &DEFAULT_COST_MODEL, + } + } +} + +impl<'a> HydraGroupingStrategy<'a> { + /// A strategy that ranks/binds via `cost_model` instead of the built-in + /// static preference order — the same customization point + /// [`crate::replacement::SketchAlgorithmStrategy::new`] already offers. + pub fn new(cost_model: &'a dyn CostModel) -> Self { + Self { cost_model } + } + + /// Every legal `SharedMultiSubpopulation` candidate for `target` — empty + /// when `target` isn't a bindable aggregate, has no subpopulation + /// concept, or its intent's candidate summary families have no Hydra + /// variant modeled. + fn hydra_candidates(&self, target: &TargetSubDAG<'_>) -> Vec { + let QueryExpr::Aggregate { reduction, .. } = target.root.as_ref() else { + return Vec::new(); + }; + if !has_subpopulations(reduction) { + return Vec::new(); + } + let Some(intent) = bindable_intent(target.root) else { + return Vec::new(); + }; + summary_candidates(intent) + .iter() + .filter_map(|kind| hydra_kind_for(kind).map(|hydra_kind| (kind.clone(), hydra_kind))) + .filter_map(|(sketch_kind, hydra_kind)| { + self.build_candidate(target.root, intent, sketch_kind, hydra_kind) + }) + .collect() + } + + /// Find the already-ranked candidate [`Implementation::Sketch`] matching + /// `sketch_kind` among [`implementations_for_with`]'s exhaustive list for + /// `intent`, bind `root` to that exact, already-decided candidate via + /// [`crate::replacement::construct_summary`] (no steering/forcing — see + /// the module docs' "No `ForceSketchKind`-style steering"), then swap the + /// resulting `SummaryAgg`'s `grouping` field from the default + /// `PerSubpopulationInstance` to + /// `SharedMultiSubpopulation { kind: hydra_kind, .. }` — reusing the + /// entire bind decision procedure (schema derivation, column resolution, + /// readout construction) unchanged, patching only the one field this + /// axis owns. + fn build_candidate( + &self, + root: &Rc, + intent: &AggIntent, + sketch_kind: SketchAlgorithm, + hydra_kind: HydraKind, + ) -> Option { + let implementation = implementations_for_with(intent, self.cost_model) + .into_iter() + .find(|candidate| { + matches!(candidate, Implementation::Sketch(kind) if *kind.algorithm() == sketch_kind) + })?; + let node = construct_summary(root, implementation, self.cost_model).ok()?; + let per_subpopulation_params = per_subpopulation_sketch_params(&node)?; + let params = default_hydra_params(hydra_kind.clone(), &per_subpopulation_params)?; + let grouping = GroupingStrategy::SharedMultiSubpopulation { + kind: hydra_kind.clone(), + params, + }; + + let patched = with_grouping(node, grouping); + Some(ReplacementSubDAG { + replacement: Replacement::Summary(patched), + rationale: format!( + "{} realizes as a shared {hydra_kind:?} structure over {sketch_kind:?} \ + serving every subpopulation of this grouped aggregate, instead of one \ + {sketch_kind:?} instance per distinct `by` key — legal because this \ + aggregate has a non-empty subpopulation concept and {sketch_kind:?} has a \ + modeled Hydra variant (asap_types::post_asap::hydra_kind_for); whether it's \ + *worth* the shared/independent trade-off for the actual subpopulation \ + cardinality is a CostModel's call, not this strategy's", + describe_intent(intent) + ), + }) + } +} + +impl ReplacementStrategy for HydraGroupingStrategy<'_> { + fn matches(&self, target: &TargetSubDAG<'_>) -> bool { + !self.hydra_candidates(target).is_empty() + } + + fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec { + self.hydra_candidates(target) + } +} + +/// The [`SketchParams`] a bound sketch candidate's `SummaryAgg` committed +/// to, if its family is `Sketch(_)` at all — `None` for any other bound +/// shape (an exact accumulator, a pass-through, or a family with no +/// `SketchParams`), never expected here in practice since `hydra_candidates` +/// only calls this for a `sketch_kind` it already confirmed has a +/// `HydraKind` via `hydra_kind_for`, but degrading to "no candidate" rather +/// than panicking keeps this as conservative as the rest of this module. +/// +/// Deliberately returns the *whole* [`SketchParams`], not one scalar field +/// pulled out of it (an earlier version of this function assumed +/// `SketchParams::Kll { k }` specifically and returned a bare `k: u32`). +/// This axis is a "sketch of sketches" framework: the inner sketch a Hydra +/// structure wraps isn't always KLL, and each [`HydraKind`] variant needs +/// its own inner sketch's own knobs — `HydraCms`/`HydraCountSketch` need +/// (`width`, `depth`), not a `k`. [`default_hydra_params`] is what actually +/// destructures the right variant for `kind`; this function's only job is +/// to find whatever `SketchParams` the bind decision already committed to +/// and hand the whole thing over unchanged. +fn per_subpopulation_sketch_params(node: &SummaryNode) -> Option { + match &node.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => { + per_subpopulation_sketch_params(summary_input) + } + SummaryExpr::SummaryAgg { + family: SummaryFamilyType::Sketch(kind, _), + .. + } => Some(kind.params().clone()), + _ => None, + } +} + +/// Rebuild `node`, replacing its `SummaryAgg`'s `grouping` field with +/// `grouping` — patching the one field this axis owns onto an +/// already-correctly-bound node rather than re-deriving the rest of it. +/// Recurses through a `SummaryEstimate` readout wrapper (the shape every +/// sketch candidate this module builds actually has) to reach the +/// `SummaryAgg` underneath. +fn with_grouping(node: Rc, grouping: GroupingStrategy) -> Rc { + match &node.expr { + SummaryExpr::SummaryEstimate { + summary_input, + query, + } => Rc::new(SummaryNode { + expr: SummaryExpr::SummaryEstimate { + summary_input: with_grouping(Rc::clone(summary_input), grouping), + query: query.clone(), + }, + schema: node.schema.clone(), + }), + SummaryExpr::SummaryAgg { + child, + family, + col, + reduction, + .. + } => { + let grouped_family = match family { + SummaryFamilyType::Sketch(kind, _) => { + SummaryFamilyType::Sketch(kind.clone(), grouping.clone()) + } + _ => family.clone(), + }; + let mut grouped_schema = node.schema.clone(); + for field in &mut grouped_schema.fields { + if let SummaryFamilyType::Sketch(kind, _) = &field.dtype { + field.dtype = SummaryFamilyType::Sketch(kind.clone(), grouping.clone()); + } + } + Rc::new(SummaryNode { + expr: SummaryExpr::SummaryAgg { + child: Rc::clone(child), + family: grouped_family, + col: col.clone(), + reduction: reduction.clone(), + grouping, + }, + schema: grouped_schema, + }) + } + // Never reached by this module's own callers (they only ever pass a + // node `construct_summary` just bound for a `Sketch` + // candidate, which is always `SummaryAgg` or + // `SummaryEstimate(SummaryAgg)`) — returning the node unchanged + // rather than panicking keeps this as conservative as the rest of + // the module if that ever stops holding. + _ => node, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::post_asap::HydraParams; + use asap_types::pre_asap::agg_intent::{default_cardinality, default_quantile}; + use asap_types::pre_asap::query_expr::Source; + use asap_types::pre_asap::schema::{Column, DataType, Schema}; + use asap_types::types::AccuracyTarget; + + fn metric_scan(labels: &[&str]) -> QueryExpr { + let mut columns = vec![ + Column::new("ts", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ]; + columns.extend(labels.iter().map(|n| Column::new(*n, DataType::Utf8, true))); + QueryExpr::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index(columns, 0, vec![]), + } + } + + fn agg(by: Vec, intent: AggIntent, child: QueryExpr) -> QueryExpr { + QueryExpr::Aggregate { + reduction: Reduction::by(by), + measures: vec![intent], + output_names: vec![], + having: None, + child: Rc::new(child), + } + } + + fn agg_per_entity(intent: AggIntent, child: QueryExpr) -> QueryExpr { + QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![intent], + output_names: vec![], + having: None, + child: Rc::new(child), + } + } + + // ── has_subpopulations ──────────────────────────────────────────────── + + #[test] + fn per_entity_has_no_subpopulation_concept() { + assert!(!has_subpopulations(&Reduction::PerEntity)); + } + + #[test] + fn empty_by_reduction_has_no_subpopulation_concept() { + assert!(!has_subpopulations(&Reduction::by(vec![]))); + } + + #[test] + fn non_empty_by_reduction_has_a_subpopulation_concept() { + assert!(has_subpopulations(&Reduction::by(vec![2]))); + } + + #[test] + fn without_grouping_has_a_subpopulation_concept_even_when_empty() { + use asap_types::pre_asap::query_expr::GroupKeys; + // `without([])` groups by every remaining label — a real + // subpopulation concept, unlike `by([])`'s genuine full reduction. + assert!(has_subpopulations(&Reduction::Reduce(GroupKeys::without( + vec![] + )))); + } + + // ── HydraGroupingStrategy ───────────────────────────────────────────── + + #[test] + fn matches_a_grouped_count_aggregate() { + let intent = AggIntent::Count { + accuracy: AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }, + }; + let q = Rc::new(agg(vec![2], intent, metric_scan(&["job"]))); + let target = TargetSubDAG::new(&q); + assert!(HydraGroupingStrategy::default_cost_model().matches(&target)); + } + + #[test] + fn does_not_match_an_empty_by_aggregate() { + // Global reduction — no subpopulation concept, no Hydra alternative. + let q = Rc::new(agg(vec![], default_quantile(0.99), metric_scan(&["job"]))); + let target = TargetSubDAG::new(&q); + let strategy = HydraGroupingStrategy::default_cost_model(); + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } + + #[test] + fn does_not_match_a_per_entity_aggregate() { + let q = Rc::new(agg_per_entity( + default_quantile(0.99), + metric_scan(&["job"]), + )); + let target = TargetSubDAG::new(&q); + let strategy = HydraGroupingStrategy::default_cost_model(); + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } + + #[test] + fn does_not_match_a_non_aggregate_node() { + let scan = Rc::new(metric_scan(&["job"])); + let target = TargetSubDAG::new(&scan); + assert!(!HydraGroupingStrategy::default_cost_model().matches(&target)); + } + + #[test] + fn quantile_has_no_hydra_candidate_without_a_modeled_error_bound() { + let q = Rc::new(agg(vec![2], default_quantile(0.99), metric_scan(&["job"]))); + let target = TargetSubDAG::new(&q); + let replacements = HydraGroupingStrategy::default_cost_model().replacements(&target); + assert!(replacements.is_empty(), "{replacements:?}"); + } + + #[test] + fn count_offers_hydra_candidates_for_cms_and_count_sketch() { + // summary_candidates(Count) = [Cms, CountSketch] — both are now + // mapped to a Hydra variant (and, per `HydraKind`'s own doc, both + // are the Hydra paper's actual proven construction, unlike KLL's). + let intent = AggIntent::Count { + accuracy: AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }, + }; + let q = Rc::new(agg(vec![2], intent, metric_scan(&["job"]))); + let target = TargetSubDAG::new(&q); + let replacements = HydraGroupingStrategy::default_cost_model().replacements(&target); + assert_eq!(replacements.len(), 2, "{replacements:?}"); + + for replacement in &replacements { + let Replacement::Summary(node) = &replacement.replacement else { + panic!("expected a Summary replacement"); + }; + let SummaryExpr::SummaryEstimate { summary_input, .. } = &node.expr else { + panic!("expected SummaryEstimate root, got {:?}", node.expr); + }; + let SummaryExpr::SummaryAgg { + family, grouping, .. + } = &summary_input.expr + else { + panic!("expected SummaryAgg, got {:?}", summary_input.expr); + }; + let SummaryFamilyType::Sketch(kind, state_grouping) = family else { + panic!("expected a Sketch family, got {family:?}"); + }; + assert_eq!(state_grouping, grouping); + assert!(summary_input + .schema + .fields + .iter() + .any(|field| &field.dtype == family)); + + // The Hydra params must carry over exactly the same + // (width, depth) the per-subpopulation candidate committed to — + // `default_hydra_params` generalizes over *which* inner sketch + // it's wrapping rather than assuming a KLL-shaped `k`. + match kind.algorithm() { + SketchAlgorithm::Cms => { + let SketchParams::Cms { width, depth } = kind.params() else { + panic!("expected Cms params, got {:?}", kind.params()); + }; + assert_eq!( + grouping, + &GroupingStrategy::SharedMultiSubpopulation { + kind: HydraKind::HydraCms, + params: HydraParams::HydraCms { + width: *width, + depth: *depth, + shared_rows: *depth, + shared_columns: *width, + }, + } + ); + } + SketchAlgorithm::CountSketch => { + let SketchParams::CountSketch { width, depth } = kind.params() else { + panic!("expected CountSketch params, got {:?}", kind.params()); + }; + assert_eq!( + grouping, + &GroupingStrategy::SharedMultiSubpopulation { + kind: HydraKind::HydraCountSketch, + params: HydraParams::HydraCountSketch { + width: *width, + depth: *depth, + shared_rows: *depth, + shared_columns: *width, + }, + } + ); + } + other => panic!("unexpected Hydra candidate algorithm: {other:?}"), + } + assert!(!replacement.rationale.is_empty()); + } + } + + #[test] + fn cardinality_has_no_hydra_candidate_yet() { + // summary_candidates(Cardinality) = [Hll, Theta, Kmv] — none have a + // modeled Hydra variant, so no candidate at all (not an error, just + // an empty result, same conservatism as every other strategy here). + let q = Rc::new(agg(vec![2], default_cardinality(), metric_scan(&["job"]))); + let target = TargetSubDAG::new(&q); + let strategy = HydraGroupingStrategy::default_cost_model(); + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } + + #[test] + fn exact_accuracy_target_has_no_hydra_candidate() { + // AccuracyTarget::Exact never binds a sketch at all — nothing for + // this axis to offer a shared-structure alternative to. + let intent = AggIntent::Quantile { + col: None, + q: 0.99, + accuracy: AccuracyTarget::Exact, + }; + let q = Rc::new(agg(vec![2], intent, metric_scan(&["job"]))); + let target = TargetSubDAG::new(&q); + let strategy = HydraGroupingStrategy::default_cost_model(); + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } + + #[test] + fn exact_mergeable_intent_has_no_hydra_candidate() { + // Sum's exact accumulator has no candidate summary families at all + // (summary_candidates only covers approximate-capable intents). + let q = Rc::new(agg( + vec![2], + AggIntent::Sum { col: None }, + metric_scan(&["job"]), + )); + let target = TargetSubDAG::new(&q); + let strategy = HydraGroupingStrategy::default_cost_model(); + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } + + #[test] + fn does_not_match_a_multi_intent_or_having_aggregate() { + let strategy = HydraGroupingStrategy::default_cost_model(); + + let multi = Rc::new(QueryExpr::Aggregate { + reduction: Reduction::by(vec![2]), + measures: vec![AggIntent::Sum { col: None }, AggIntent::Avg { col: None }], + output_names: vec![], + having: None, + child: Rc::new(metric_scan(&["job"])), + }); + let target = TargetSubDAG::new(&multi); + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + } + + /// A custom `CostModel` doesn't change *which* candidate is offered — + /// only which sketch candidate `implementations_for_with` itself would + /// have ranked first, and how that candidate's own params are sized — + /// same guarantee `SketchAlgorithmStrategy` makes for its own candidates. + struct PreferDDSketch; + impl CostModel for PreferDDSketch { + fn rank_candidates( + &self, + _intent: &AggIntent, + candidates: &[SketchAlgorithm], + ) -> Vec { + let mut v = candidates.to_vec(); + if let Some(pos) = v.iter().position(|k| *k == SketchAlgorithm::DDSketch) { + let dd = v.remove(pos); + v.insert(0, dd); + } + v + } + } + + #[test] + fn custom_cost_model_cannot_enable_unproven_hydra_kll() { + let q = Rc::new(agg(vec![2], default_quantile(0.99), metric_scan(&["job"]))); + let target = TargetSubDAG::new(&q); + let custom = PreferDDSketch; + let replacements = HydraGroupingStrategy::new(&custom).replacements(&target); + assert!(replacements.is_empty(), "{replacements:?}"); + } +} diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index 89fc2ab5..1863800a 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -108,6 +108,12 @@ //! 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. +//! - [`grouping`] — [`grouping::HydraGroupingStrategy`] (issue #256, part of +//! #33) is an additional `ReplacementStrategy`: the orthogonal +//! `GroupingStrategy` axis (one summary instance per `by` subpopulation +//! versus one shared Hydra-family structure serving all of them), offered +//! alongside the candidates [`replacement::SketchAlgorithmStrategy`] +//! enumerates for the same target. //! - [`rewrite`] — the "semantic-equivalent rewriting (e.g. `avg` → //! `sum`/`count`) to increase how often the [sharing/sketch] optimizations //! above apply" degree of freedom `docs/design_docs/asap_aware_mapping.md` @@ -165,6 +171,7 @@ pub mod cost_model; pub mod explanation; +pub mod grouping; pub mod replacement; pub mod rewrite; pub mod rollup; @@ -173,6 +180,7 @@ pub use cost_model::{CostModel, DefaultCostModel}; pub use explanation::{ explain_replacements, explain_replacements_with, ExplanationKind, ReplacementExplanation, }; +pub use grouping::{has_subpopulations, HydraGroupingStrategy}; pub use replacement::{ default_strategies, default_strategies_with, search_workload, search_workload_with, summary_candidates, ImplementError, Implementation, Matcher, MemoGroup, PlanSpace, RankedGroup, diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 2504fc8d..e091975c 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -264,9 +264,10 @@ use std::collections::HashMap; use asap_types::post_asap::{ - ExactKind, ExactParams, SamplingKind, SamplingParams, SketchAlgorithm, SketchKind, - SketchParams, SketchQuery as PostAsapSketchQuery, StatModelKind, StatModelParams, SummaryExpr, - SummaryFamilyType, SummaryField, SummaryNode, SummarySchema, WaveletKind, WaveletParams, + ExactKind, ExactParams, GroupingStrategy, SamplingKind, SamplingParams, SketchAlgorithm, + SketchKind, SketchParams, SketchQuery as PostAsapSketchQuery, StatModelKind, StatModelParams, + SummaryExpr, SummaryFamilyType, SummaryField, SummaryNode, SummarySchema, WaveletKind, + WaveletParams, }; use asap_types::pre_asap::agg_intent::{agg_is_mergeable, AggIntent}; use asap_types::pre_asap::cse::{share_common_subtrees, structural_hash, HashCache}; @@ -278,6 +279,7 @@ use std::rc::Rc; use thiserror::Error; use crate::cost_model::{CostModel, CseCandidate, DefaultCostModel, ShareDecision}; +use crate::grouping::HydraGroupingStrategy; use crate::rollup::RollupStrategy; /// Errors from the pre-ASAP → post-ASAP replacement/construction path @@ -540,11 +542,15 @@ pub fn accuracy_target(intent: &AggIntent) -> Option<&AccuracyTarget> { /// explicit realization is a compile error, and the coverage-matrix test pins /// each variant's category. /// -/// Module-private: [`SketchAlgorithmStrategy::replacements`] is the only -/// caller — a caller outside this module has no use for the bare -/// `Implementation` list on its own, only for the bound -/// [`ReplacementSubDAG`]s that strategy produces from it. -fn implementations_for_with(intent: &AggIntent, cost_model: &dyn CostModel) -> Vec { +/// `pub(crate)`: [`SketchAlgorithmStrategy::replacements`] is this module's +/// own caller; `grouping::HydraGroupingStrategy` (issue #256) is the one +/// caller outside it, needing the exact same already-ranked candidate list +/// to find the `Implementation::Sketch` matching the Hydra-eligible kind it +/// is building a candidate for. +pub(crate) fn implementations_for_with( + intent: &AggIntent, + cost_model: &dyn CostModel, +) -> Vec { match intent { // ── Approximate-capable intents — the AccuracyTarget decides ──────── AggIntent::Quantile { accuracy, .. } @@ -1047,7 +1053,10 @@ fn describe_implementation(intent: &AggIntent, implementation: &Implementation) /// counterpart of its own: it reads a candidate's `rationale` — built from /// this text — straight off [`ReplacementSubDAG`], rather than re-describing /// the same intent a second time. -fn describe_intent(intent: &AggIntent) -> String { +/// +/// `pub(crate)`: `grouping::HydraGroupingStrategy` (issue #256) reuses this +/// for its own rationale strings, for the same reason. +pub(crate) fn describe_intent(intent: &AggIntent) -> String { match intent { AggIntent::Quantile { q, .. } => format!("quantile(q={q})"), AggIntent::Cardinality { .. } => "cardinality (distinct count)".to_string(), @@ -1155,7 +1164,15 @@ pub fn bindable_intent(node: &QueryExpr) -> Option<&AggIntent> { /// child goes back through [`realize_child`] (fresh candidate /// enumeration, not a forced pick), so choosing one candidate for a target /// never leaks into that target's own nested aggregates. -fn construct_summary( +/// +/// `pub(crate)`: `grouping::HydraGroupingStrategy` (issue #256) is the one +/// caller outside this module — the same first-class, +/// one-candidate-at-a-time primitive [`SketchAlgorithmStrategy`] itself +/// calls once per candidate, reused rather than duplicated so a Hydra +/// candidate gets exactly the same schema derivation/column +/// resolution/readout construction as every other candidate, patching only +/// the `grouping` field this axis owns. +pub(crate) fn construct_summary( expr: &QueryExpr, implementation: Implementation, cost_model: &dyn CostModel, @@ -1193,7 +1210,10 @@ fn summary_family(implementation: Implementation) -> Option<(SummaryFamilyType, Implementation::ExactAggregate { kind, params } => { (SummaryFamilyType::ExactAggregate(kind, params), false) } - Implementation::Sketch(kind) => (SummaryFamilyType::Sketch(kind), true), + Implementation::Sketch(kind) => ( + SummaryFamilyType::Sketch(kind, GroupingStrategy::default()), + true, + ), Implementation::Sample { kind, params } => (SummaryFamilyType::Sample(kind, params), true), Implementation::Wavelet { kind, params } => { (SummaryFamilyType::Wavelet(kind, params), true) @@ -1248,6 +1268,7 @@ fn construct_summary_agg( family, col, reduction: reduction.clone(), + grouping: GroupingStrategy::default(), }, schema: state_schema, }); @@ -1597,7 +1618,8 @@ impl PlanSpace { /// /// Ranking itself is decided entirely by [`rank_group`] before /// [`RankedGroup::costs`] is ever computed — pairing each candidate with - /// [`CostModel::estimate_cost`]'s own number is an additive annotation + /// [`CostModel::grouping_state_cost`] for grouping alternatives, or + /// [`CostModel::estimate_cost`] otherwise, is an additive annotation /// for a caller that wants to *display* a cost (e.g. a /// DAG-visualization view), not a second ranking signal, so plugging in /// a `CostModel` whose `estimate_cost` disagrees with its own @@ -1614,7 +1636,11 @@ impl PlanSpace { let target = TargetSubDAG::with_consumer_count(&group.target, group.consumer_count); let costs = candidates .iter() - .map(|c| cost_model.estimate_cost(c, &target)) + .map(|c| { + cost_model + .grouping_state_cost(c, &target) + .map_or_else(|| cost_model.estimate_cost(c, &target), |cost| cost.0) + }) .collect(); RankedGroup { target: &group.target, @@ -1634,7 +1660,8 @@ pub struct RankedGroup<'a> { pub target: &'a Rc, pub consumer_count: usize, pub candidates: Vec<&'a ReplacementSubDAG>, - /// `costs[i]` is `candidates[i]`'s own [`CostModel::estimate_cost`] + /// `costs[i]` is `candidates[i]`'s own grouping-state cost when available, + /// and its [`CostModel::estimate_cost`] otherwise /// estimate — aligned index-for-index with `candidates`, one number per /// candidate, for a caller that wants an actual `f64` next to each /// candidate (e.g. "candidate A costs ≈ X, candidate B costs ≈ Y") and @@ -1674,7 +1701,44 @@ fn rank_group<'a>(group: &'a MemoGroup, cost_model: &dyn CostModel) -> Vec<&'a R return ranked; } - // Shape 2: `SketchAlgorithmStrategy`'s sketch-family candidates (every + // Shape 2: independent and Hydra grouping alternatives for the same + // sketch algorithms. When deployment statistics provide a subpopulation + // estimate, compare N independent states with the shared grid directly. + let target = TargetSubDAG::with_consumer_count(&group.target, group.consumer_count); + let has_hydra = ranked.iter().any(|candidate| { + let Replacement::Summary(node) = &candidate.replacement else { + return false; + }; + summary_grouping(node).is_some_and(|grouping| { + matches!(grouping, GroupingStrategy::SharedMultiSubpopulation { .. }) + }) + }); + let grouping_costs: Option> = if has_hydra { + ranked + .iter() + .map(|candidate| { + cost_model + .grouping_state_cost(candidate, &target) + .map(|cost| cost.0) + }) + .collect() + } else { + None + }; + if let Some(costs) = grouping_costs { + let by_ptr: HashMap<*const ReplacementSubDAG, f64> = ranked + .iter() + .zip(costs) + .map(|(candidate, cost)| (*candidate as *const ReplacementSubDAG, cost)) + .collect(); + ranked.sort_by(|a, b| { + by_ptr[&(*a as *const ReplacementSubDAG)] + .total_cmp(&by_ptr[&(*b as *const ReplacementSubDAG)]) + }); + return ranked; + } + + // Shape 3: `SketchAlgorithmStrategy`'s sketch-family candidates (every // candidate is a `Summary` that realizes a `SketchAlgorithm`) — rank via // `CostModel::rank_candidates`, the same hook `implementations_for_with` // itself consults. @@ -1706,7 +1770,6 @@ fn rank_group<'a>(group: &'a MemoGroup, cost_model: &dyn CostModel) -> Vec<&'a R // types, so compare the numeric estimates the CostModel exposes for that // purpose. `total_cmp` gives deterministic placement to a model's NaN // placeholders without dropping any candidate. - let target = TargetSubDAG::with_consumer_count(&group.target, group.consumer_count); ranked.sort_by(|a, b| { cost_model .estimate_cost(a, &target) @@ -1768,13 +1831,23 @@ fn sketch_kind_of(node: &SummaryNode) -> Option { match &node.expr { SummaryExpr::SummaryEstimate { summary_input, .. } => sketch_kind_of(summary_input), SummaryExpr::SummaryAgg { - family: SummaryFamilyType::Sketch(kind), + family: SummaryFamilyType::Sketch(kind, _), .. } => Some(kind.algorithm().clone()), _ => None, } } +/// The grouping strategy used by a bound summary candidate, unwrapping its +/// readout node when necessary. +fn summary_grouping(node: &SummaryNode) -> Option<&GroupingStrategy> { + match &node.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => summary_grouping(summary_input), + SummaryExpr::SummaryAgg { grouping, .. } => Some(grouping), + _ => None, + } +} + // ── default_strategies ────────────────────────────────────────────────── /// The context-free strategies [`search_workload`] runs in the built-in @@ -1789,6 +1862,7 @@ fn sketch_kind_of(node: &SummaryNode) -> Option { pub fn default_strategies() -> Vec> { vec![ Box::new(SketchAlgorithmStrategy::default_cost_model()), + Box::new(HydraGroupingStrategy::default_cost_model()), Box::new(SharedSubtreeStrategy), ] } @@ -1801,6 +1875,7 @@ pub fn default_strategies_with<'a>( ) -> Vec> { vec![ Box::new(SketchAlgorithmStrategy::new(cost_model)), + Box::new(HydraGroupingStrategy::new(cost_model)), Box::new(SharedSubtreeStrategy), ] } @@ -2831,7 +2906,9 @@ mod tests { summary_family_algorithm(summary_input) } asap_types::post_asap::SummaryExpr::SummaryAgg { family, .. } => match family { - asap_types::post_asap::SummaryFamilyType::Sketch(kind) => kind.algorithm().clone(), + asap_types::post_asap::SummaryFamilyType::Sketch(kind, _) => { + kind.algorithm().clone() + } other => panic!("expected a Sketch family, got {other:?}"), }, other => panic!("expected SummaryAgg/SummaryEstimate, got {other:?}"), @@ -3009,8 +3086,14 @@ mod tests { // ── discovery + MEMO shape ─────────────────────────────────────────── #[test] - fn single_bindable_aggregate_gets_a_group_with_every_sketch_candidate() { - let root = Rc::new(agg(vec![2], default_quantile(0.99), metric_scan(&["job"]))); + fn single_bindable_aggregate_gets_every_sketch_and_grouping_candidate() { + let intent = AggIntent::Count { + accuracy: AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }, + }; + let root = Rc::new(agg(vec![2], intent, metric_scan(&["job"]))); let space = search_workload(vec![("q", root)]); // One group for the Aggregate, one for its Scan child. @@ -3023,14 +3106,37 @@ mod tests { assert_eq!(agg_group.consumer_count, 1); assert_eq!( agg_group.candidates.len(), - 2, - "quantile has 2 summary_candidates entries: {:?}", + 4, + "grouped approximate count has independent and Hydra CMS/CountSketch candidates: {:?}", agg_group.candidates ); assert!(agg_group .candidates .iter() .all(|c| matches!(c.replacement, Replacement::Summary(_)))); + assert_eq!( + agg_group + .candidates + .iter() + .filter(|candidate| { + let Replacement::Summary(node) = &candidate.replacement else { + return false; + }; + let SummaryExpr::SummaryEstimate { summary_input, .. } = &node.expr else { + return false; + }; + matches!( + &summary_input.expr, + SummaryExpr::SummaryAgg { + grouping: GroupingStrategy::SharedMultiSubpopulation { .. }, + .. + } + ) + }) + .count(), + 2, + "the default workload search must register the Hydra grouping strategy" + ); let scan_group = space .groups() @@ -3355,6 +3461,62 @@ mod tests { assert_eq!(first_kind, Some(SketchAlgorithm::DDSketch)); } + #[test] + fn grouping_cost_prefers_hydra_only_for_high_subpopulation_cardinality() { + struct EstimatedSubpopulations(usize); + + impl CostModel for EstimatedSubpopulations { + fn rank_candidates( + &self, + _intent: &AggIntent, + candidates: &[SketchAlgorithm], + ) -> Vec { + candidates.to_vec() + } + + fn estimated_subpopulation_count(&self, _target: &QueryExpr) -> Option { + Some(self.0) + } + } + + fn first_grouping(estimated_count: usize) -> GroupingStrategy { + let model = EstimatedSubpopulations(estimated_count); + let intent = AggIntent::Count { + accuracy: AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }, + }; + let root = Rc::new(agg( + vec![2, 3], + intent, + metric_scan(&["tenant_id", "endpoint"]), + )); + let strategies = default_strategies_with(&model); + let space = search_workload_with(vec![("tenant_endpoint_count", root)], &strategies); + let ranked = space.cost_sorted(&model); + let aggregate = ranked + .iter() + .find(|group| matches!(group.target.as_ref(), QueryExpr::Aggregate { .. })) + .expect("aggregate group"); + let Replacement::Summary(node) = &aggregate.candidates[0].replacement else { + panic!("grouping candidate must be a summary") + }; + summary_grouping(node) + .expect("bound summary grouping") + .clone() + } + + assert!(matches!( + first_grouping(10_000), + GroupingStrategy::SharedMultiSubpopulation { .. } + )); + assert_eq!( + first_grouping(10), + GroupingStrategy::PerSubpopulationInstance + ); + } + /// [`RankedGroup::costs`] is a per-candidate annotation, aligned /// index-for-index with `candidates` — each entry must equal what /// calling [`CostModel::estimate_cost`] directly on that same candidate @@ -3520,26 +3682,27 @@ mod tests { family, col, reduction, + .. } = &summary_input.expr else { panic!("expected SummaryAgg, got {:?}", summary_input.expr); }; assert_eq!( family, - &SummaryFamilyType::Sketch(SketchKind::new( - SketchAlgorithm::Kll, - SketchParams::Kll { k: 200 } - )) + &SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + GroupingStrategy::default() + ) ); assert_eq!(col, &ColumnRef::SampleValue); assert_eq!(reduction, &ReductionTy::by(vec![2])); // SummaryAgg edge: the state column carries the committed family. assert_eq!( field(&summary_input.schema, "quantile_0_99").dtype, - SummaryFamilyType::Sketch(SketchKind::new( - SketchAlgorithm::Kll, - SketchParams::Kll { k: 200 } - )) + SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + GroupingStrategy::default() + ) ); assert!(matches!(child.expr, SummaryExpr::KeepPreAsap(ref e) if matches!(**e, QueryExpr::Scan { .. }))); @@ -3580,7 +3743,7 @@ mod tests { }; assert!(matches!( family, - SummaryFamilyType::Sketch(kind) if kind.algorithm() == &SketchAlgorithm::Kll + SummaryFamilyType::Sketch(kind, _) if kind.algorithm() == &SketchAlgorithm::Kll )); // With `PreferDDSketchViaCostModel`: DDSketch instead, same query. @@ -3593,10 +3756,13 @@ mod tests { }; assert_eq!( family, - &SummaryFamilyType::Sketch(SketchKind::new( - SketchAlgorithm::DDSketch, - SketchParams::DDSketch { alpha: 0.01 } - )) + &SummaryFamilyType::Sketch( + SketchKind::new( + SketchAlgorithm::DDSketch, + SketchParams::DDSketch { alpha: 0.01 } + ), + GroupingStrategy::default() + ) ); } @@ -3690,13 +3856,16 @@ mod tests { }; assert_eq!( family, - &SummaryFamilyType::Sketch(SketchKind::new( - SketchAlgorithm::CountSketch, - SketchParams::CountSketch { - width: 256, - depth: 4 - } - )) + &SummaryFamilyType::Sketch( + SketchKind::new( + SketchAlgorithm::CountSketch, + SketchParams::CountSketch { + width: 256, + depth: 4 + } + ), + GroupingStrategy::default() + ) ); } @@ -3819,7 +3988,7 @@ mod tests { }; assert!(matches!( family, - SummaryFamilyType::Sketch(kind) if kind.algorithm() == &SketchAlgorithm::Kll + SummaryFamilyType::Sketch(kind, _) if kind.algorithm() == &SketchAlgorithm::Kll )); let SummaryExpr::SummaryAgg { family: inner_family, @@ -3962,7 +4131,7 @@ mod tests { assert!(matches!( &summary_input.expr, SummaryExpr::SummaryAgg { - family: SummaryFamilyType::Sketch(kind), + family: SummaryFamilyType::Sketch(kind, _), .. } if kind.algorithm() == &SketchAlgorithm::CmsWithHeap )); diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs index 1c810406..d0f48da8 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -14,8 +14,8 @@ use asap_aware_mapping::{ }; use asap_frontend_promql::lower_promql; use asap_types::post_asap::{ - ExactKind, ExactParams, SketchAlgorithm, SketchKind, SketchParams, SketchQuery, SummaryExpr, - SummaryFamilyType, SummaryNode, SummarySchema, + ExactKind, ExactParams, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, + SketchQuery, SummaryExpr, SummaryFamilyType, SummaryNode, SummarySchema, }; use asap_types::pre_asap::expr_ir::ColumnRef; use asap_types::pre_asap::query_expr::{QueryExpr, Reduction}; @@ -98,16 +98,17 @@ fn promql_quantile_of_rate_binds_kll_over_rate_accumulator() { family, col, reduction, + .. } = &summary_input.expr else { panic!("expected SummaryAgg, got {:?}", summary_input.expr); }; assert_eq!( family, - &SummaryFamilyType::Sketch(SketchKind::new( - SketchAlgorithm::Kll, - SketchParams::Kll { k: 200 } - )) + &SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + GroupingStrategy::default() + ) ); assert_eq!(col, &ColumnRef::SampleValue); assert_eq!( @@ -117,10 +118,10 @@ fn promql_quantile_of_rate_binds_kll_over_rate_accumulator() { ); assert_eq!( dtype(&summary_input.schema, "quantile_0_99"), - &SummaryFamilyType::Sketch(SketchKind::new( - SketchAlgorithm::Kll, - SketchParams::Kll { k: 200 } - )) + &SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + GroupingStrategy::default() + ) ); // The rate: exact counter-reset-aware accumulator, per-series (labels diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs index 9fde0f17..42564b04 100644 --- a/crates/integration-tests/tests/sql_to_post_asap.rs +++ b/crates/integration-tests/tests/sql_to_post_asap.rs @@ -38,8 +38,8 @@ use asap_aware_mapping::{ }; use asap_frontend_sql::{lower_sql, SqlCatalog}; use asap_types::post_asap::{ - ExactKind, ExactParams, SketchAlgorithm, SketchKind, SketchParams, SketchQuery, SummaryExpr, - SummaryFamilyType, SummaryNode, SummarySchema, + ExactKind, ExactParams, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, + SketchQuery, SummaryExpr, SummaryFamilyType, SummaryNode, SummarySchema, }; use asap_types::pre_asap::expr_ir::ColumnRef; use asap_types::pre_asap::query_expr::{QueryExpr, Reduction}; @@ -186,16 +186,17 @@ async fn sql_quantile_binds_kll_sketch_over_named_column() { family, col, reduction, + .. } = &summary_input.expr else { panic!("expected SummaryAgg, got {:?}", summary_input.expr); }; assert_eq!( family, - &SummaryFamilyType::Sketch(SketchKind::new( - SketchAlgorithm::Kll, - SketchParams::Kll { k: 200 } - )) + &SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + GroupingStrategy::default() + ) ); assert_eq!( col, @@ -212,10 +213,10 @@ async fn sql_quantile_binds_kll_sketch_over_named_column() { ); assert_eq!( summary_input.schema.fields[0].dtype, - SummaryFamilyType::Sketch(SketchKind::new( - SketchAlgorithm::Kll, - SketchParams::Kll { k: 200 } - )) + SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + GroupingStrategy::default() + ) ); let SummaryExpr::KeepPreAsap(kept_leaf) = &child.expr else { @@ -273,10 +274,10 @@ async fn sql_count_distinct_binds_hll_sketch_over_named_column() { }; assert_eq!( family, - &SummaryFamilyType::Sketch(SketchKind::new( - SketchAlgorithm::Hll, - SketchParams::Hll { precision: 14 } - )) + &SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Hll, SketchParams::Hll { precision: 14 }), + GroupingStrategy::default() + ) ); assert_eq!( col, diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index 1c9958e0..1111f5d2 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -1,7 +1,7 @@ use std::rc::Rc; use super::schema::{SummaryFamilyType, SummarySchema}; -use super::sketch::SketchQuery; +use super::sketch::{GroupingStrategy, SketchQuery}; use crate::pre_asap::{ColumnRef, QueryExpr, Reduction}; // ── Post-ASAP DAG node ─────────────────────────────────────────────────────── @@ -59,6 +59,21 @@ pub enum SummaryExpr { /// two collapsed to the same ambiguous `by: []` before this field /// existed (issue #163). reduction: Reduction, + /// How this aggregation's summary state is physically instantiated + /// across `reduction`'s subpopulations — one independent instance + /// per `by` key (today's only behavior, and this field's default), + /// or one shared Hydra-family structure serving all of them (issue + /// #256). Lives here, next to `reduction`, for planning, and is also + /// encoded in sketch-valued `family`/output-schema state so merges + /// can reject incompatible layouts. `reduction` is the field that + /// carries the `by` keys this axis's legality depends on (a + /// `SharedMultiSubpopulation` choice only makes sense when + /// `reduction` actually has a subpopulation concept — see + /// `asap_aware_mapping::grouping`'s module docs for the legality + /// rules). Every existing producer of a `SummaryAgg` sets this to + /// `GroupingStrategy::PerSubpopulationInstance` (its `Default`), + /// so no existing behavior changes. + grouping: GroupingStrategy, }, /// Summary-aware join (KMV / theta for join-cardinality; join-sample for diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs index 0953cf8a..5809eb75 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -13,9 +13,19 @@ //! one-pair-per-family shape: it nests a third level, [`sketch::SketchKind`] //! (quantile/cardinality/frequency/top-k), which itself carries the //! committed [`sketch::SketchAlgorithm`] and [`sketch::SketchParams`] — -//! `SummaryFamilyType::Sketch(SketchKind)`, not a flat `(kind, params)` pair +//! `SummaryFamilyType::Sketch(SketchKind, GroupingStrategy)`, not a flat +//! `(kind, params)` pair //! — because `Sketch` is the one family with more than one algorithm per //! purpose today; no other family needs that extra level yet. +//! +//! A second, orthogonal axis lives here too: [`sketch::GroupingStrategy`] +//! (issue #256) — *how many* physical instances of a chosen family/kind +//! exist across a grouped aggregate's `by` subpopulations +//! (`PerSubpopulationInstance`, today's only behavior, vs. +//! `SharedMultiSubpopulation`/Hydra — see [`sketch::HydraKind`]/ +//! [`sketch::HydraParams`]), carried on [`expr::SummaryExpr::SummaryAgg`] +//! alongside `reduction` and on sketch-valued edge types +//! — see `asap_aware_mapping::grouping`'s module docs for why. pub mod expr; pub mod query_time; @@ -29,7 +39,7 @@ pub use query_time::{ }; pub use schema::{SummaryFamilyType, SummaryField, SummarySchema}; pub use sketch::{ - ExactKind, ExactParams, SamplingKind, SamplingParams, SketchAlgorithm, SketchCategory, - SketchKind, SketchParams, SketchQuery, StatModelKind, StatModelParams, WaveletKind, - WaveletParams, + default_hydra_params, hydra_kind_for, ExactKind, ExactParams, GroupingStrategy, HydraKind, + HydraParams, SamplingKind, SamplingParams, SketchAlgorithm, SketchCategory, SketchKind, + SketchParams, SketchQuery, StatModelKind, StatModelParams, WaveletKind, WaveletParams, }; diff --git a/crates/types/src/post_asap/schema.rs b/crates/types/src/post_asap/schema.rs index 0f51b42b..f8e8b249 100644 --- a/crates/types/src/post_asap/schema.rs +++ b/crates/types/src/post_asap/schema.rs @@ -1,6 +1,6 @@ use super::sketch::{ - ExactKind, ExactParams, SamplingKind, SamplingParams, SketchKind, StatModelKind, - StatModelParams, WaveletKind, WaveletParams, + ExactKind, ExactParams, GroupingStrategy, SamplingKind, SamplingParams, SketchKind, + StatModelKind, StatModelParams, WaveletKind, WaveletParams, }; use crate::pre_asap::DataType; @@ -11,8 +11,9 @@ use crate::pre_asap::DataType; /// edges that carry partial summary state between a `SummaryAgg` and a /// downstream `SummaryEstimate` or `SummaryMerge`. /// -/// Every non-`Plain` variant carries `(kind, params)` from that family's own -/// pair of types, so the type system can reject merges of incompatible +/// Every non-`Plain` variant carries the physical state identity required by +/// that family (`Sketch` additionally carries its grouping layout), so the +/// type system can reject merges of incompatible /// summaries at plan construction time — a `SummaryMerge` over /// `Sketch(Kll, …)` and `Sketch(Cms, …)` inputs is a plan-time error, and a /// `Sketch(…)` can never be confused for a `Sample(…)` even though both are @@ -28,10 +29,10 @@ pub enum SummaryFamilyType { ExactAggregate(ExactKind, ExactParams), /// Approximate sketch state (KLL/CMS/HLL/…), read out via a /// `SummaryEstimate`. A [`SketchKind`] already carries the concrete - /// algorithm and params committed to, not just its category — a bound - /// node needs to know it's specifically KLL, not merely "some quantile - /// sketch". - Sketch(SketchKind), + /// algorithm, params, and grouping layout committed to, not just its + /// category — a bound node needs to know it's specifically independent + /// KLL or shared Hydra-backed CMS, not merely "some sketch". + Sketch(SketchKind, GroupingStrategy), /// Sampling-based summary state (a retained row subset). Sample(SamplingKind, SamplingParams), /// Wavelet-transform summary state (a coefficient vector). diff --git a/crates/types/src/post_asap/sketch.rs b/crates/types/src/post_asap/sketch.rs index a627fffe..e3a4a8b8 100644 --- a/crates/types/src/post_asap/sketch.rs +++ b/crates/types/src/post_asap/sketch.rs @@ -267,6 +267,253 @@ pub enum StatModelParams { }, } +// ── Grouping strategy: per-subpopulation vs. shared multi-subpopulation +// (Hydra) — issue #256 ────────────────────────────────────────────────── +// +// A grouped aggregate (`GROUP BY city, quantile(...)`) has always implicitly +// built one independent summary instance per distinct `by` key — there was +// no type anywhere expressing that as a *choice* rather than a foregone +// conclusion. Hydra (Manousis et al., VLDB 2022 — see `HydraKind`'s own doc +// for the full citation and which variants are its actual proven +// construction) is the alternative: one shared structure serving every +// subpopulation instead of N independent instances, trading memory/build +// cost against per-subpopulation isolation. +// +// This axis is deliberately modeled here, alongside `SketchKind`/ +// `SamplingKind`/`WaveletKind`/`StatModelKind`, rather than as a new +// `SketchKind` (or `SamplingKind`, etc.) entry: it is orthogonal to *which* +// family/kind answers an intent — any family could in principle grow its own +// per-subpopulation vs. shared-multi-subpopulation variant, so it is a +// second, independent axis, not a member of any one family's own kind +// vocabulary. See `asap_aware_mapping::grouping`'s module docs for where this +// axis actually plugs into the post-ASAP IR and the legality rules gating +// when `SharedMultiSubpopulation` is offered as a candidate at all. + +/// A shared-multi-subpopulation summary family — one physical structure +/// serving every subpopulation of a grouped aggregate instead of one +/// independent instance per distinct `by` key. Named after Hydra (Manousis, +/// Cheng, Ben Basat, Liu, Sekar. "Enabling Efficient and General +/// Subpopulation Analytics in Multidimensional Data Streams." VLDB 2022). +/// +/// Orthogonal to [`SketchKind`]/[`SamplingKind`]/[`WaveletKind`]/ +/// [`StatModelKind`] the same way [`GroupingStrategy`] as a whole is +/// orthogonal to them. +/// +/// ## Which variants are the paper's own proven construction, and which aren't +/// +/// Hydra's accuracy proof (paper §4.5, Theorem 2) is over one specific +/// construction: hash each subpopulation into one of a shared w×r grid of +/// **linear, mergeable frequency-vector sketches** — the paper's own +/// heavy-hitter substrate is Count-Sketch (§4.3); Count-Min Sketch is the +/// same collision algebra — and bound the noise a colliding subpopulation's +/// estimate picks up from the others sharing its cell. +/// [`HydraKind::HydraCms`]/[`HydraKind::HydraCountSketch`] are exactly that +/// construction over this crate's existing `SketchAlgorithm::Cms`/ +/// `CountSketch`, so the paper's bound applies to them directly. +/// +/// [`HydraKind::HydraKll`] is **not** an instance of that proven +/// construction: KLL is an order-statistics sketch, not a linear frequency +/// vector, and has no analogous "sum the colliding contributions, bound the +/// noise" algebra. The paper is explicit that its own construction cannot +/// serve quantiles at all (§4.3: "A statistic that cannot directly be +/// estimated by Hydra-sketch is quantiles."). `HydraKll` remains available +/// as an explicit experimental IR value, but [`hydra_kind_for`] does not +/// expose it to semantics-preserving replacement search: no error bound is +/// modeled for it (see [`HydraParams::HydraKll`]'s own doc). Enable automatic +/// selection only alongside an actual proof and error model. +/// +/// Not yet modeled: `HydraUnivMon`. The paper's own named "Hydra-sketch" is +/// really the universal-sketch composition (L layers of Count-Sketch plus a +/// heavy-hitter heap, Theorems 1+2 combined) estimating entropy/L1-norm/ +/// L2-norm/cardinality/frequency-moments as one instance. That needs new +/// `AggIntent`/category vocabulary this crate doesn't have yet (no +/// `Entropy`/`L1Norm`/`L2Norm` intents) and is deliberately out of scope +/// here — see issue #256's follow-up. +#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub enum HydraKind { + /// Hydra over a KLL-family quantile sketch. See this type's own doc: + /// **not** an instance of the paper's proven construction — the paper + /// excludes quantiles from Hydra-sketch entirely. Kept only as an + /// explicit experimental representation; automatic replacement search + /// must not treat it as equivalent to independent instances. + HydraKll, + /// Hydra over Count-Min Sketch: a direct instance of the paper's proven + /// w×r shared-grid construction (§4.2/§4.5) — CMS is exactly the + /// linear frequency-vector substrate Theorem 2 is proved over. + HydraCms, + /// Hydra over Count-Sketch — the paper's own heavy-hitter substrate + /// (§4.3). A direct instance of the same proven construction as + /// `HydraCms`, with Count-Sketch's balanced/zero-mean-error trade + /// instead of CMS's one-sided bias. + HydraCountSketch, +} + +/// Parameters for a [`HydraKind`] instance. Unlike a plain [`SketchParams`] +/// (sized purely for one instance's own accuracy), a Hydra structure needs +/// both the per-subpopulation accuracy target it emulates *and* how big the +/// one shared structure itself is — the latter is the whole memory/accuracy +/// trade Hydra makes, and has no equivalent for an instance built +/// independently per subpopulation. One variant per [`HydraKind`] because +/// the knobs a "sketch of sketches" needs are specific to the inner +/// sketch's own parameter shape — a `HydraCms` instance is sized in +/// (`width`, `depth`), a `HydraKll` instance in `k`; there is no single knob +/// set general enough to cover every inner sketch type. +#[derive(Debug, Clone, PartialEq)] +pub enum HydraParams { + /// See [`HydraKind::HydraKll`]: kept for legality, **no accuracy bound + /// is modeled for this variant**. `k`/`shared_buckets` give a bound + /// sketch a concrete size to build against; unlike `HydraCms`/ + /// `HydraCountSketch`'s fields, they are not backed by the paper's + /// Theorem 2. + HydraKll { + /// Per-subpopulation accuracy knob — same meaning as + /// `SketchParams::Kll`'s own `k`, for the per-subpopulation logical + /// view this shared structure emulates. + k: u32, + /// Sizing knob for the one physical structure shared across every + /// subpopulation. Correctly sizing this against an estimated + /// subpopulation cardinality is a cost-model concern — out of scope + /// for the legality axis this type lives on (see + /// `asap_aware_mapping::grouping`'s module docs) — so this is + /// deliberately not derived from any cardinality estimate here. + shared_buckets: u32, + }, + /// Hydra over Count-Min Sketch — the paper's proven w×r + /// shared-grid construction (§4.2, Theorem 2). `width`/`depth` are the + /// per-subpopulation CMS knobs (mirroring `SketchParams::Cms`'s own + /// fields); `shared_rows`/`shared_columns` are the paper's `r`/`w` — the + /// redundant hash rows and shared-bucket width of the one physical grid + /// every subpopulation is hashed into. + HydraCms { + width: u32, + depth: u32, + /// The paper's `r`: redundant, pairwise-independent hash rows — + /// query time takes the median across rows to tighten the failure + /// probability (Theorem 2's `δ` term). + shared_rows: u32, + /// The paper's `w`: shared buckets per row that colliding + /// subpopulations share — the memory/accuracy knob Theorem 2's `ε` + /// term depends on. Sizing this against an estimated subpopulation + /// cardinality is a cost-model concern, deliberately out of scope + /// for the legality axis this type lives on. + shared_columns: u32, + }, + /// Hydra over Count-Sketch — same shape, and the same Theorem 2, as + /// `HydraCms`, over `SketchParams::CountSketch`'s knobs instead. + HydraCountSketch { + width: u32, + depth: u32, + shared_rows: u32, + shared_columns: u32, + }, +} + +/// Which [`HydraKind`] (if any) provides a shared-multi-subpopulation +/// variant of a plain per-subpopulation [`SketchAlgorithm`]. `None` means +/// this axis's scope stops at legality: not every `SketchAlgorithm` has a +/// Hydra wrapper modeled yet — extend alongside `HydraKind` as more are +/// added. See [`HydraKind`]'s own doc for which of the mapped kinds are the +/// paper's own proven construction. Unproven extensions such as `HydraKll` +/// deliberately return `None`: replacement search may only expose variants +/// with a modeled error guarantee. +pub fn hydra_kind_for(algorithm: &SketchAlgorithm) -> Option { + match algorithm { + SketchAlgorithm::Cms => Some(HydraKind::HydraCms), + SketchAlgorithm::CountSketch => Some(HydraKind::HydraCountSketch), + _ => None, + } +} + +/// Default [`HydraParams`] for `kind`, carrying over `per_subpopulation_params` +/// — the [`SketchParams`] a plain, independent-per-subpopulation instance of +/// the same algorithm would have used — unchanged into the corresponding +/// `HydraParams` fields. `None` when `per_subpopulation_params` doesn't +/// belong to the [`SketchAlgorithm`] `kind` wraps: a caller bug, since +/// [`hydra_kind_for`] and the algorithm a `SketchParams` came from must +/// agree; callers that got both from the same already-ranked +/// `Implementation` (as `asap_aware_mapping::grouping` does) cannot hit +/// this. +/// +/// This function is generic over which inner sketch type `kind` wraps +/// precisely because [`SketchParams`] already is: it destructures whichever +/// variant matches `kind` rather than assuming a single scalar knob (e.g. a +/// bare `k: u32`) that only KLL happens to have — a Hydra "sketch of +/// sketches" is a framework over *any* mergeable inner sketch, and `HydraCms`/ +/// `HydraCountSketch`'s (`width`, `depth`) pairs are just as much a +/// per-subpopulation accuracy knob as `HydraKll`'s `k`. +/// +/// `shared_buckets`/`shared_rows`/`shared_columns` are all sized to the same +/// per-subpopulation value as a placeholder pending real cost-model-driven +/// sizing (the paper's own `r`/`w`, §4.6) — see each field's own doc for why +/// that's deliberately out of scope here. +pub fn default_hydra_params( + kind: HydraKind, + per_subpopulation_params: &SketchParams, +) -> Option { + match (kind, per_subpopulation_params) { + (HydraKind::HydraKll, SketchParams::Kll { k }) => Some(HydraParams::HydraKll { + k: *k, + shared_buckets: *k, + }), + (HydraKind::HydraCms, SketchParams::Cms { width, depth }) => Some(HydraParams::HydraCms { + width: *width, + depth: *depth, + shared_rows: *depth, + shared_columns: *width, + }), + (HydraKind::HydraCountSketch, SketchParams::CountSketch { width, depth }) => { + Some(HydraParams::HydraCountSketch { + width: *width, + depth: *depth, + shared_rows: *depth, + shared_columns: *width, + }) + } + _ => None, + } +} + +/// How a grouped aggregate's summary state is physically instantiated +/// across its `by` subpopulations — orthogonal to *which* +/// `SketchKind`/`SamplingKind`/`WaveletKind`/`StatModelKind` answers the +/// intent (that choice lives alongside it on `SummaryFamilyType`). Lives here, +/// alongside `SketchKind`/`SketchParams` etc., rather than on any of those +/// enums themselves, for exactly the reason explained in this section's +/// module docs above. +/// +/// Carried both on `SummaryExpr::SummaryAgg` (where planning consults it) +/// and on sketch-valued `SummaryFamilyType` edges (where it prevents +/// incompatible shared and independent physical states from type-checking +/// as merge-compatible). +#[derive(Debug, Clone, PartialEq)] +pub enum GroupingStrategy { + /// One independent summary instance per distinct `by` key — today's + /// only (implicit) behavior, and this type's `Default` (below), so + /// every existing caller that never chose this axis explicitly keeps + /// observing exactly the same behavior it always has. + PerSubpopulationInstance, + /// One shared structure serving every subpopulation (Hydra and its + /// per-family variants — see [`HydraKind`]), trading per-subpopulation + /// isolation for shared memory/build cost. + /// + /// Named `SharedMultiSubpopulation`, not `SharedMultiTenant`: this + /// shares across a *query's own* subpopulations/group-by keys, not + /// across tenants in a deployment-isolation sense — a different, and + /// unrelated, kind of "sharing". + SharedMultiSubpopulation { + kind: HydraKind, + params: HydraParams, + }, +} + +impl Default for GroupingStrategy { + /// [`GroupingStrategy::PerSubpopulationInstance`] — see that variant's + /// own doc for why this is the only sound default. + fn default() -> Self { + Self::PerSubpopulationInstance + } +} + // ── Sketch read-out queries ─────────────────────────────────────────────────── /// What to extract from a built summary. Carried by `SummaryEstimate`. @@ -308,4 +555,92 @@ mod tests { fn sketch_kind_rejects_params_from_another_algorithm() { SketchKind::new(SketchAlgorithm::Kll, SketchParams::Hll { precision: 14 }); } + + #[test] + fn grouping_strategy_default_is_per_subpopulation_instance() { + // Existing/default behavior must stay `PerSubpopulationInstance` — + // this must not change any existing test's observed behavior. + assert_eq!( + GroupingStrategy::default(), + GroupingStrategy::PerSubpopulationInstance + ); + } + + #[test] + fn hydra_kind_for_only_maps_algorithms_with_modeled_error_bounds() { + assert_eq!(hydra_kind_for(&SketchAlgorithm::Kll), None); + assert_eq!( + hydra_kind_for(&SketchAlgorithm::Cms), + Some(HydraKind::HydraCms) + ); + assert_eq!( + hydra_kind_for(&SketchAlgorithm::CountSketch), + Some(HydraKind::HydraCountSketch) + ); + // Every other `SketchAlgorithm` has no Hydra variant modeled yet — + // a deliberate, documented scope limit, not an oversight. + for algorithm in [ + SketchAlgorithm::Hll, + SketchAlgorithm::DDSketch, + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::Kmv, + SketchAlgorithm::Theta, + SketchAlgorithm::CountSketchWithHeap, + ] { + assert_eq!(hydra_kind_for(&algorithm), None, "{algorithm:?}"); + } + } + + #[test] + fn default_hydra_params_carries_over_per_subpopulation_params_by_kind() { + assert_eq!( + default_hydra_params(HydraKind::HydraKll, &SketchParams::Kll { k: 200 }), + Some(HydraParams::HydraKll { + k: 200, + shared_buckets: 200, + }) + ); + assert_eq!( + default_hydra_params( + HydraKind::HydraCms, + &SketchParams::Cms { + width: 2048, + depth: 4, + } + ), + Some(HydraParams::HydraCms { + width: 2048, + depth: 4, + shared_rows: 4, + shared_columns: 2048, + }) + ); + assert_eq!( + default_hydra_params( + HydraKind::HydraCountSketch, + &SketchParams::CountSketch { + width: 2048, + depth: 4, + } + ), + Some(HydraParams::HydraCountSketch { + width: 2048, + depth: 4, + shared_rows: 4, + shared_columns: 2048, + }) + ); + } + + #[test] + fn default_hydra_params_rejects_a_mismatched_kind_and_params_pair() { + // A `HydraCms` kind paired with KLL params (or vice versa) is a + // caller bug — `hydra_kind_for` and the algorithm a `SketchParams` + // came from must agree. Degrades to `None` rather than panicking, + // matching this module's conservative stance elsewhere. + assert_eq!( + default_hydra_params(HydraKind::HydraCms, &SketchParams::Kll { k: 200 }), + None + ); + } }