diff --git a/crates/executor/src/capability.rs b/crates/executor/src/capability.rs index 9d066730..89975bcc 100644 --- a/crates/executor/src/capability.rs +++ b/crates/executor/src/capability.rs @@ -22,7 +22,7 @@ pub fn validate_summary_kernel( grouping: &GroupingStrategy, ) -> Result<(), String> { if grouping != &GroupingStrategy::PerSubpopulationInstance { - return Err("shared summary grouping has no registered kernel".into()); + return validate_hydra_cms(family, input, grouping); } let keyed = match family { SummaryFamilyType::ExactAggregate(kind, params) => { @@ -111,6 +111,76 @@ pub fn validate_summary_kernel( Ok(()) } +/// The only shared grouping with a kernel: Hydra over Count-Min. Each update +/// adds a unit or non-negative column weight for one item of one group. +fn validate_hydra_cms( + family: &SummaryFamilyType, + input: &SummaryUpdate, + grouping: &GroupingStrategy, +) -> Result<(), String> { + use planner_types::ir::schema::{SummaryInputExpr, WeightDomain}; + hydra_cms_shape(family, grouping)?; + if !matches!(input.item, Some(SummaryInputExpr::Column(_))) { + return Err("HydraCms requires one item column".into()); + } + if !matches!(input.weight_domain, WeightDomain::NonNegative { .. }) + || !matches!( + input.weight, + SummaryInputExpr::Constant(1.0) | SummaryInputExpr::Column(_) + ) + { + return Err("HydraCms requires a unit or non-negative column weight".into()); + } + Ok(()) +} + +/// `(shared_rows, shared_columns, width, depth)` of a HydraCms state. Its +/// family is the per-group Count-Min it emulates, with the Hydra's own width +/// and depth. +pub(crate) fn hydra_cms_shape( + family: &SummaryFamilyType, + grouping: &GroupingStrategy, +) -> Result<(usize, usize, usize, usize), String> { + use planner_types::ir::schema::{HydraKind, HydraParams}; + let GroupingStrategy::SharedMultiSubpopulation { + kind: HydraKind::HydraCms, + params: + HydraParams::HydraCms { + width, + depth, + shared_rows, + shared_columns, + }, + } = grouping + else { + return Err("shared summary grouping has no registered kernel".into()); + }; + let SummaryFamilyType::Sketch(kind, layout) = family else { + return Err("HydraCms requires a Count-Min family".into()); + }; + if layout != grouping { + return Err("Planner family and operator grouping disagree".into()); + } + if kind.algorithm() != &SketchAlgorithm::Cms + || kind.params() + != &(SketchParams::Cms { + width: *width, + depth: *depth, + }) + { + return Err("HydraCms family must be Count-Min with the Hydra width and depth".into()); + } + if !valid_matrix(*width, *depth) || !valid_matrix(*shared_columns, *shared_rows) { + return Err("invalid HydraCms dimensions".into()); + } + Ok(( + *shared_rows as usize, + *shared_columns as usize, + *width as usize, + *depth as usize, + )) +} + fn valid_matrix(width: u32, depth: u32) -> bool { // Construction uses the kernel's native row hashing, so no encoded-size // limit applies here. @@ -142,9 +212,15 @@ pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error> { use planner_types::ir::schema::SketchAlgorithm as A; if let SummaryFamilyType::Sketch(kind, grouping) = family { // Plain Count-Min is native as stored state only: it merges and reads - // its bare count, but the DAG does not build it from rows. + // its bare count, but the DAG does not build it from rows. HydraCms + // is built by the shared summary operator. if let (A::Cms, SketchParams::Cms { width, depth }) = (kind.algorithm(), kind.params()) { - return if valid_matrix(*width, *depth) && grouping == &Default::default() { + if grouping != &GroupingStrategy::PerSubpopulationInstance { + return hydra_cms_shape(family, grouping) + .map(|_| ()) + .map_err(Error::Invalid); + } + return if valid_matrix(*width, *depth) { Ok(()) } else { Err(Error::Invalid( @@ -211,6 +287,13 @@ pub fn validate_sketch_evaluation( | (A::UnivMon, SketchStatistic::Cardinality) | (A::UnivMon, SketchStatistic::FrequencyL2) | (A::UnivMon, SketchStatistic::FrequencyEntropy) => true, + // HydraCms answers a group's item frequency as well as its total. + (A::Cms, SketchStatistic::PointCount { .. }) + if matches!(family, SummaryFamilyType::Sketch(_, grouping) + if grouping != &GroupingStrategy::PerSubpopulationInstance) => + { + true + } // Only count intents read a Count-Min bare count, and their // updates have unit weight; the evaluation is typed Int64 on that basis. (A::Cms, _) => bare_count, diff --git a/crates/executor/src/operators/mod.rs b/crates/executor/src/operators/mod.rs index 5f521ac3..1859fbdb 100644 --- a/crates/executor/src/operators/mod.rs +++ b/crates/executor/src/operators/mod.rs @@ -147,6 +147,13 @@ enum Kind { items: Vec, groups: Vec, }, + /// One shared state for all groups (HydraCms); `weight: None` is a unit count. + SharedSummaryBuild { + family: SummaryFamilyType, + item: usize, + weight: Option, + groups: Vec, + }, KeyedEvaluation { state: usize, k: usize, @@ -291,6 +298,7 @@ impl PhysicalOperator for Operator { | Kind::SemiJoin { .. } | Kind::SummaryBuild { .. } | Kind::KeyedSummaryBuild { .. } + | Kind::SharedSummaryBuild { .. } | Kind::SummaryMerge { .. } | Kind::VectorToScalar { .. } ) @@ -342,7 +350,9 @@ impl PhysicalOperator for Operator { Kind::Window { .. } => "WindowAggregate", Kind::SemiJoin { .. } => "SemiJoin", Kind::Join { .. } => "RelationalJoin", - Kind::SummaryBuild { .. } | Kind::KeyedSummaryBuild { .. } => "SummaryAgg", + Kind::SummaryBuild { .. } + | Kind::KeyedSummaryBuild { .. } + | Kind::SharedSummaryBuild { .. } => "SummaryAgg", Kind::KeyedEvaluation { .. } => "SummaryEstimate", Kind::SummaryMerge { .. } => "SummaryMerge", Kind::Evaluation { .. } => "SummaryEvaluation", @@ -397,6 +407,7 @@ impl PhysicalOperator for Operator { Kind::SummaryBuild { .. } | Kind::Evaluation { .. } | Kind::KeyedSummaryBuild { .. } + | Kind::SharedSummaryBuild { .. } | Kind::KeyedEvaluation { .. } => summary::execute(self, inputs, context), } } diff --git a/crates/executor/src/operators/summary/mod.rs b/crates/executor/src/operators/summary/mod.rs index 9f4e9cf2..18b6dbe5 100644 --- a/crates/executor/src/operators/summary/mod.rs +++ b/crates/executor/src/operators/summary/mod.rs @@ -60,6 +60,49 @@ impl Operator { output: schema(fields), }) } + /// One HydraCms state shared by every group, emitted as one row per group + /// that reads it. `weight: None` counts each row once. + pub fn shared_summary_build( + input: SchemaRef, + family: SummaryFamilyType, + item: usize, + weight: Option, + groups: Vec, + ) -> Result { + crate::values::validate_family(&family)?; + crate::factory::create_hydra_cms(&family).map_err(Error::Invalid)?; + validate_groups(&input, &groups)?; + if !matches!( + plain(&input, item)?.0, + DataType::Utf8 | DataType::Int64 | DataType::Bool + ) { + return Err(invalid("HydraCms items must be Utf8, Int64 or Bool")); + } + if weight.is_some_and(|weight| !matches!(plain(&input, weight), Ok((DataType::Float64, _)))) + { + return Err(invalid("HydraCms weight must be Float64")); + } + let mut fields = groups + .iter() + .map(|&i| input.fields[i].clone()) + .collect::>(); + fields.push(SummaryField { + name: "state".into(), + dtype: family.clone(), + nullable: false, + table: None, + }); + Ok(Self { + kind: Kind::SharedSummaryBuild { + family, + item, + weight, + groups, + }, + inputs: vec![input], + output: schema(fields), + }) + } pub fn keyed_evaluation( input: SchemaRef, state: usize, @@ -264,6 +307,18 @@ pub(super) fn execute<'a>( ) }) .boxed_local()), + Kind::SharedSummaryBuild { + family, + item, + weight, + groups, + } => Ok(futures::stream::once(async move { + Batch::try_new( + output, + build_shared_summary(input, family, *item, *weight, groups, &context).await?, + ) + }) + .boxed_local()), Kind::KeyedEvaluation { state, k } => Ok(input .map(move |batch| { let batch = batch?; @@ -610,3 +665,78 @@ async fn build_keyed_summary( }) .collect()) } + +async fn build_shared_summary( + mut input: Input<'_, Batch>, + family: &SummaryFamilyType, + item: usize, + weight: Option, + groups: &[usize], + context: &RunContext, +) -> Result>, Error> { + use crate::summary_kernels::HydraCmsGroup; + let mut grid = crate::factory::create_hydra_cms(family).map_err(Error::Invalid)?; + let grid_bytes = grid.approx_memory_bytes(); + let mut memory = context.reserve(grid_bytes)?; + let mut retained = grid_bytes; + let mut work = Cooperative::new(context); + // Group labels and the group's Hydra subpopulation name, an injective + // encoding of its typed key. + let mut seen = BTreeMap::>, (Vec, String)>::new(); + while let Some(batch) = input.next().await { + let batch = batch?; + for row in batch.rows() { + work.checkpoint().await?; + let key = group_key(row, groups)?; + if !seen.contains_key(&key) { + let name = key + .iter() + .map(|part| part.iter().map(|b| format!("{b:02x}")).collect::()) + .collect::>() + .join(","); + let labels = groups.iter().map(|&i| row[i].clone()).collect::>(); + retained += + key_bytes(&key) + labels.iter().map(Value::bytes).sum::() + name.len(); + memory.resize(retained)?; + seen.insert(key.clone(), (labels, name)); + } + let count = match weight.map(|column| &row[column]) { + None => 1, + // SQL aggregates ignore NULL weights while retaining the group. + Some(Value::Null) => continue, + Some(Value::Float64(w)) + if *w >= 0.0 && w.fract() == 0.0 && *w <= f64::from(i32::MAX) => + { + *w as i32 + } + Some(_) => { + return Err(Error::Operator( + "HydraCms weights must be non-negative integers".into(), + )) + } + }; + let item = match &row[item] { + Value::Utf8(v) => v.to_string(), + Value::Int64(v) => v.to_string(), + Value::Bool(v) => v.to_string(), + _ => return Err(Error::Operator("HydraCms item must be non-null".into())), + }; + grid.update(&seen[&key].1, &item, count) + .map_err(|e| Error::Operator(e.to_string()))?; + } + } + let grid = Arc::new(grid); + Ok(seen + .into_values() + .map(|(mut labels, group)| { + labels.push(Value::Summary { + family: family.clone(), + state: Arc::new(HydraCmsGroup { + grid: grid.clone(), + group, + }), + }); + labels + }) + .collect()) +} diff --git a/crates/executor/src/operators/unchecked.rs b/crates/executor/src/operators/unchecked.rs index 46e72eac..98310430 100644 --- a/crates/executor/src/operators/unchecked.rs +++ b/crates/executor/src/operators/unchecked.rs @@ -162,6 +162,12 @@ impl TryFrom for Operator { items, groups, } => Operator::keyed_summary_build(input(0)?, family, value, items, groups)?, + Kind::SharedSummaryBuild { + family, + item, + weight, + groups, + } => Operator::shared_summary_build(input(0)?, family, item, weight, groups)?, Kind::KeyedEvaluation { state, k } => { Operator::keyed_evaluation(input(0)?, state, k, output.clone())? } diff --git a/crates/executor/src/physical_planner/mod.rs b/crates/executor/src/physical_planner/mod.rs index 9f9d689e..090cef9f 100644 --- a/crates/executor/src/physical_planner/mod.rs +++ b/crates/executor/src/physical_planner/mod.rs @@ -923,6 +923,27 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result Some(named_column(input, weight)?), + _ => None, + }; + return Operator::shared_summary_build( + input.clone(), + family.clone(), + named_column(input, item)?, + weight, + groups(input, keys)?, + ); + } if let Some(item) = &update.item { let PlannerReduction::Reduce(keys) = reduction else { return Err(invalid("keyed summary requires explicit partitions")); diff --git a/crates/executor/src/summary_kernels/factory.rs b/crates/executor/src/summary_kernels/factory.rs index 016c18c5..fe414ff0 100644 --- a/crates/executor/src/summary_kernels/factory.rs +++ b/crates/executor/src/summary_kernels/factory.rs @@ -2,7 +2,7 @@ use crate::summary_kernels::hll_sketch::HllSketchAccumulator; use crate::summary_kernels::univmon::UnivMonAccumulator; use crate::summary_kernels::{ CountMinSketchAccumulator, CountMinSketchWithHeapAccumulator, CountSketchAccumulator, - CountSketchWithHeapAccumulator, DDSketchAccumulator, DatasketchesKLLAccumulator, + CountSketchWithHeapAccumulator, DDSketchAccumulator, DatasketchesKLLAccumulator, HydraCms, HydraKllSketchAccumulator, }; use crate::{AggregateCore, KeyByLabelValues}; @@ -563,7 +563,9 @@ pub fn create_planner_accumulator( crate::capability::validate_summary_kernel(family, input, grouping)?; use planner_types::ir::schema::GroupingStrategy; if grouping != &GroupingStrategy::PerSubpopulationInstance { - return Err("shared summary grouping requires a supported Planner Hydra kernel".into()); + return Err( + "shared grouping has one state for all groups; build it with `create_hydra_cms`".into(), + ); } if matches!(family, SummaryFamilyType::ExactAggregate(..)) { return Ok(Box::new(PlannerExactUpdater { @@ -647,6 +649,18 @@ pub fn create_planner_accumulator( Ok(updater) } +/// Construct the one state a `SharedMultiSubpopulation { kind: HydraCms }` +/// family shares across its groups. The update contract is checked with the +/// plan by `validate_summary_kernel`. +pub fn create_hydra_cms(family: &SummaryFamilyType) -> Result { + let SummaryFamilyType::Sketch(_, grouping) = family else { + return Err("HydraCms requires a Count-Min family".into()); + }; + let (shared_rows, shared_columns, width, depth) = + crate::capability::hydra_cms_shape(family, grouping)?; + HydraCms::new(shared_rows, shared_columns, width, depth).map_err(|e| e.to_string()) +} + struct PlannerExactUpdater { acc: crate::summary_kernels::exact::ExactAccumulator, } @@ -818,4 +832,47 @@ mod planner_parameter_regression { assert_eq!(dims, (3, 128), "{algorithm:?}"); } } + + // HydraCms builds one shared grid shaped by its Hydra params; it is never + // a per-group updater, and its family must be the matching Count-Min. + #[test] + fn hydra_cms_builds_one_shared_grid() { + use planner_types::ir::schema::{GroupingStrategy, HydraKind, HydraParams}; + let grouping = |width| GroupingStrategy::SharedMultiSubpopulation { + kind: HydraKind::HydraCms, + params: HydraParams::HydraCms { + width, + depth: 3, + shared_rows: 5, + shared_columns: 32, + }, + }; + let family = |width| { + SummaryFamilyType::Sketch( + SketchKind::new( + SketchAlgorithm::Cms, + SketchParams::Cms { + width: 128, + depth: 3, + }, + ), + grouping(width), + ) + }; + assert_eq!( + create_hydra_cms(&family(128)).unwrap().shape(), + (5, 32, 128, 3) + ); + assert!(create_hydra_cms(&family(64)).is_err()); + let update = SummaryUpdate { + item: Some(SummaryInputExpr::Column( + planner_types::ir::scalar::ColumnRef::Named("host".into()), + )), + weight: SummaryInputExpr::Constant(1.0), + weight_domain: planner_types::ir::schema::WeightDomain::NonNegative { + proof: planner_types::ir::schema::NonNegativeWeightProof::UnitCount, + }, + }; + assert!(create_planner_accumulator(&family(128), &update, &grouping(128)).is_err()); + } } diff --git a/crates/executor/src/summary_kernels/hydra_cms.rs b/crates/executor/src/summary_kernels/hydra_cms.rs new file mode 100644 index 00000000..635f5feb --- /dev/null +++ b/crates/executor/src/summary_kernels/hydra_cms.rs @@ -0,0 +1,319 @@ +//! Hydra over Count-Min Sketch (`HydraKind::HydraCms`) over `asap_sketchlib::Hydra`. +//! +//! One `shared_rows × shared_columns` grid of Count-Min cells serves every +//! group: a group hashes to one cell per row, and that cell's Count-Min +//! counts the group's items. Colliding groups share a cell, so both readouts +//! only overestimate (weights are non-negative). +use crate::{AggregateCore, KernelError}; +use asap_sketchlib::{ + hash_for_matrix_seeded, input::HydraCounter, CountMin, DataInput, FastPath, Hydra, Vector2D, + HYDRA_SEED, +}; +use planner_types::ir::schema::SketchStatistic; +use std::sync::Arc; + +/// The single Hydra key column: a group's whole label tuple is one +/// subpopulation, so no record fans out into sub-groupings. +const GROUP: &str = "group"; + +/// One shared grid for all groups, sized by `HydraParams::HydraCms`. +#[derive(Debug, Clone)] +pub struct HydraCms { + inner: Hydra, + width: usize, + depth: usize, + /// Total update weight. Every counter is at most this, so keeping it + /// within `i32` keeps sketchlib's `i32` counters from overflowing. + total: i64, +} + +impl HydraCms { + pub fn new( + shared_rows: usize, + shared_columns: usize, + width: usize, + depth: usize, + ) -> Result { + if [shared_rows, shared_columns, width, depth].contains(&0) { + return Err("HydraCms dimensions must be positive".into()); + } + let cell = HydraCounter::CM(CountMin::, FastPath>::with_dimensions( + depth, width, + )); + Ok(Self { + inner: Hydra::with_schema(shared_rows, shared_columns, [GROUP], cell)?, + width, + depth, + total: 0, + }) + } + + /// `(shared_rows, shared_columns, width, depth)`. + pub fn shape(&self) -> (usize, usize, usize, usize) { + ( + self.inner.row_num, + self.inner.col_num, + self.width, + self.depth, + ) + } + + /// Add `weight` occurrences of `item` to `group`. + pub fn update(&mut self, group: &str, item: &str, weight: i32) -> Result<(), KernelError> { + if weight < 0 { + return Err("HydraCms weights must be non-negative".into()); + } + let total = self.total + i64::from(weight); + if total > i64::from(i32::MAX) { + return Err("HydraCms total weight exceeds its i32 counters".into()); + } + if weight > 0 { + self.inner + .update(&[group], &DataInput::Str(item), Some(weight))?; + } + self.total = total; + Ok(()) + } + + /// Estimated frequency of `item` within `group`: Hydra's median over rows + /// of each cell's Count-Min estimate (the paper's Theorem 2 estimator). + pub fn point(&self, group: &str, item: &str) -> Result { + Ok(self + .inner + .query_frequency(&[Some(group)], &DataInput::Str(item))?) + } + + /// Estimated total weight of `group`: the minimum over rows of the mass in + /// the group's cell. Every row's cell holds the group's whole mass plus + /// colliding groups', so the minimum is never below the true total, is at + /// most the median Theorem 2 bounds, and is an exact integer. + pub fn group_count(&self, group: &str) -> f64 { + let hash = hash_for_matrix_seeded( + HYDRA_SEED, + self.inner.row_num, + self.inner.col_num, + &DataInput::Str(&subkey(group)), + ); + self.inner + .sketches + .fast_query_min_with_key(&hash, &(), |cell, _, _, _| match cell { + HydraCounter::CM(cms) => cms + .as_storage() + .row_slice(0) + .iter() + .map(|&c| i64::from(c)) + .sum::(), + _ => unreachable!("HydraCms cells are Count-Min"), + }) as f64 + } + + /// Merge two grids built over disjoint inputs. Count-Min cells add + /// counter-wise under the same hashes, so the result equals the grid built + /// from the union of the inputs. + pub fn merge(&self, other: &Self) -> Result { + // sketchlib's Count-Min merge panics on a shape mismatch. + if self.shape() != other.shape() { + return Err("HydraCms merge requires identical dimensions".into()); + } + let total = self.total + other.total; + if total > i64::from(i32::MAX) { + return Err("HydraCms total weight exceeds its i32 counters".into()); + } + let mut merged = self.clone(); + merged.inner.merge(&other.inner)?; + merged.total = total; + Ok(merged) + } + + pub fn approx_memory_bytes(&self) -> usize { + let (rows, columns, width, depth) = self.shape(); + std::mem::size_of::() + rows * columns * (64 + width * depth * 4) + } +} + +/// sketchlib's canonical subkey for `{GROUP = group}` (`label:value`, with +/// `\`, `:` and `;` escaped). `Hydra::update` hashes this string; the group +/// total reads the same cells, so the encodings must agree. The kernel tests +/// compare group totals with exact counts and fail if they drift apart. +fn subkey(group: &str) -> String { + let mut out = format!("{GROUP}:"); + for ch in group.chars() { + if matches!(ch, '\\' | ':' | ';') { + out.push('\\'); + } + out.push(ch); + } + out +} + +/// One group's row state: the shared grid plus the group it reads. A grouped +/// SummaryAgg emits one row per group, all pointing at the same grid. +#[derive(Debug, Clone)] +pub struct HydraCmsGroup { + pub grid: Arc, + pub group: String, +} + +impl AggregateCore for HydraCmsGroup { + fn clone_boxed_core(&self) -> Box { + Box::new(self.clone()) + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + fn as_any_mut(&mut self) -> &mut dyn std::any::Any { + self + } + + fn merge_with(&self, other: &dyn AggregateCore) -> Result, KernelError> { + let other = other + .as_any() + .downcast_ref::() + .ok_or("HydraCms merges only with HydraCms")?; + if self.group != other.group { + return Err("HydraCms merge requires the same group".into()); + } + Ok(Box::new(Self { + grid: Arc::new(self.grid.merge(&other.grid)?), + group: self.group.clone(), + })) + } + + /// A bare point count reads the group total; a valued one reads that + /// item's frequency within the group. + fn estimate(&self, query: &SketchStatistic) -> Result { + match query { + SketchStatistic::PointCount { value: None, .. } => { + Ok(self.grid.group_count(&self.group)) + } + SketchStatistic::PointCount { + value: Some(item), .. + } => self.grid.point(&self.group, item), + _ => Err(format!("{query:?} is not supported by HydraCms").into()), + } + } + + /// The whole shared grid: groups of one build share it, so this + /// over-reserves, which is the safe direction. + fn approx_memory_bytes(&self) -> usize { + self.grid.approx_memory_bytes() + self.group.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const ROWS: [(&str, &str, i32); 9] = [ + ("api", "checkout", 3), + ("api", "checkout", 2), + ("api", "search", 1), + ("api", "a:b;c", 4), + ("batch", "checkout", 7), + ("batch", "export", 1), + ("db;x:y", "checkout", 1), + ("db;x:y", "vacuum", 6), + ("db;x:y", "vacuum", 2), + ]; + + fn exact_point(group: &str, item: &str) -> f64 { + ROWS.iter() + .filter(|(g, i, _)| *g == group && *i == item) + .map(|(_, _, w)| f64::from(*w)) + .sum() + } + + fn exact_count(group: &str) -> f64 { + ROWS.iter() + .filter(|(g, _, _)| *g == group) + .map(|(_, _, w)| f64::from(*w)) + .sum() + } + + fn build(rows: &[(&str, &str, i32)], shared_columns: usize) -> HydraCms { + let mut grid = HydraCms::new(3, shared_columns, 64, 3).unwrap(); + for (group, item, weight) in rows { + grid.update(group, item, *weight).unwrap(); + } + grid + } + + // Group totals and per-group item counts stay within the CMS bound of + // exact values; total N = 27, so the additive slack is e * N / 64 per level. + #[test] + fn group_and_point_estimates_within_cms_bound() { + let grid = build(&ROWS, 64); + let n = f64::from(ROWS.iter().map(|r| r.2).sum::()); + let slack = std::f64::consts::E * n / 64.0 * 2.0; + for group in ["api", "batch", "db;x:y"] { + let estimate = grid.group_count(group); + let exact = exact_count(group); + assert!(estimate >= exact && estimate <= exact + slack, "{group}"); + for item in ["checkout", "search", "export", "vacuum", "a:b;c"] { + let estimate = grid.point(group, item).unwrap(); + let exact = exact_point(group, item); + assert!( + estimate >= exact && estimate <= exact + slack, + "{group}/{item}: {estimate} vs {exact}" + ); + } + } + // An unseen group reads (near) zero rather than another group's cell. + assert!(grid.group_count("unseen") <= slack); + } + + // With one shared column every group collides, so each total reads the + // whole stream's mass: never below the exact value. + #[test] + fn collisions_only_overestimate() { + let grid = build(&ROWS, 1); + assert_eq!(grid.group_count("api"), 27.0); + assert!(grid.point("batch", "checkout").unwrap() >= exact_point("batch", "checkout")); + } + + // Merging grids over a split stream equals the grid over the whole stream. + #[test] + fn merge_equals_union_build() { + let (left, right) = ROWS.split_at(4); + let merged = build(left, 8).merge(&build(right, 8)).unwrap(); + let whole = build(&ROWS, 8); + for group in ["api", "batch", "db;x:y"] { + assert_eq!(merged.group_count(group), whole.group_count(group)); + assert_eq!( + merged.point(group, "checkout").unwrap(), + whole.point(group, "checkout").unwrap() + ); + } + assert!(build(left, 8).merge(&build(right, 16)).is_err()); + } + + // Group views answer only their own group and merge only with it. + #[test] + fn group_view_reads_and_merges_one_group() { + let grid = Arc::new(build(&ROWS, 64)); + let view = |group: &str| HydraCmsGroup { + grid: grid.clone(), + group: group.into(), + }; + let count = SketchStatistic::PointCount { + key: planner_types::ir::scalar::ColumnRef::SampleValue, + value: None, + }; + assert_eq!(view("batch").estimate(&count).unwrap(), 8.0); + let merged = view("batch").merge_with(&view("batch")).unwrap(); + assert_eq!(merged.estimate(&count).unwrap(), 16.0); + assert!(view("api").merge_with(&view("batch")).is_err()); + assert!(view("api").estimate(&SketchStatistic::Cardinality).is_err()); + } + + // Updates that could overflow the i32 counters are rejected. + #[test] + fn rejects_negative_and_overflowing_weights() { + let mut grid = HydraCms::new(1, 1, 1, 1).unwrap(); + assert!(grid.update("g", "x", -1).is_err()); + grid.update("g", "x", i32::MAX).unwrap(); + assert!(grid.update("g", "x", 1).is_err()); + assert!(grid.merge(&grid).is_err()); + } +} diff --git a/crates/executor/src/summary_kernels/mod.rs b/crates/executor/src/summary_kernels/mod.rs index 478e75d0..095f82f2 100644 --- a/crates/executor/src/summary_kernels/mod.rs +++ b/crates/executor/src/summary_kernels/mod.rs @@ -7,6 +7,7 @@ pub mod datasketches_kll; pub mod dd_sketch; pub mod exact; pub mod hll_sketch; +pub mod hydra_cms; pub mod hydra_kll; pub mod increase; pub mod univmon; @@ -18,6 +19,7 @@ pub use count_sketch_with_heap::*; pub use datasketches_kll::*; pub use dd_sketch::*; pub use hll_sketch::*; +pub use hydra_cms::*; pub use hydra_kll::*; pub use increase::*; diff --git a/crates/executor/src/values.rs b/crates/executor/src/values.rs index 6f2bc76b..4e17066a 100644 --- a/crates/executor/src/values.rs +++ b/crates/executor/src/values.rs @@ -286,6 +286,17 @@ fn validate_state(family: &SummaryFamilyType, state: &dyn AggregateCore) -> Resu }) } + SummaryFamilyType::Sketch(_, grouping) + if grouping + != &planner_types::ir::schema::GroupingStrategy::PerSubpopulationInstance => + { + let shape = + crate::capability::hydra_cms_shape(family, grouping).map_err(Error::Invalid)?; + state + .as_any() + .downcast_ref::() + .is_some_and(|s| s.grid.shape() == shape) + } SummaryFamilyType::ExactAggregate(..) => state .as_any() .downcast_ref::() diff --git a/crates/executor/tests/hydra_cms_execution.rs b/crates/executor/tests/hydra_cms_execution.rs new file mode 100644 index 00000000..d004e81e --- /dev/null +++ b/crates/executor/tests/hydra_cms_execution.rs @@ -0,0 +1,323 @@ +//! A Hydra-grouped Count-Min SummaryAgg compiles through the physical planner +//! and answers per-group counts and item point counts (#580, W7). +use asap_executor::{ + dag::{ + operators::Operator, + values::{Batch, SchemaRef, Value}, + Limits, RunContext, Scope, + }, + physical_planner::{compile_node, CompiledPhysicalDAG, InputContract, Source}, +}; +use futures::{executor::block_on, StreamExt}; +use planner_types::ir::export::{ + LogicalASAPNodeId, PhysicalASAPDAGNode, PhysicalASAPOperatorPayload as Payload, +}; +use planner_types::ir::operator::{GroupKeys, Reduction}; +use planner_types::ir::properties::ExecutionDataState; +use planner_types::ir::scalar::ColumnRef; +use planner_types::ir::schema::*; +use std::{collections::BTreeMap, sync::Arc}; + +const WIDTH: u32 = 64; +const SHARED_COLUMNS: u32 = 64; + +/// `(job, service, weight)` rows; jobs are the groups, services the items. +const ROWS: [(&str, &str, f64); 9] = [ + ("api", "checkout", 3.), + ("api", "checkout", 2.), + ("api", "search", 1.), + ("api", "auth", 4.), + ("batch", "checkout", 7.), + ("batch", "export", 1.), + ("db", "checkout", 1.), + ("db", "vacuum", 6.), + ("db", "vacuum", 2.), +]; + +fn grouping() -> GroupingStrategy { + GroupingStrategy::SharedMultiSubpopulation { + kind: HydraKind::HydraCms, + params: HydraParams::HydraCms { + width: WIDTH, + depth: 3, + shared_rows: 3, + shared_columns: SHARED_COLUMNS, + }, + } +} + +fn family() -> FieldDataType { + FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::Cms, + SketchParams::Cms { + width: WIDTH, + depth: 3, + }, + ), + grouping(), + ) +} + +fn schema(fields: Vec>) -> Schema { + Schema { + fields, + unique_keys: vec![], + closed: false, + time_index: None, + } +} + +fn node(id: u32, payload: Payload, output: Schema) -> PhysicalASAPDAGNode { + PhysicalASAPDAGNode { + id: LogicalASAPNodeId(id), + payload, + output_state: ExecutionDataState::QUERY_ROWS, + output_schema: output, + guarantee: None, + coverage: None, + } +} + +fn input_schema() -> SchemaRef { + Arc::new(schema(vec![ + Field::plain("job", DataType::Utf8, false), + Field::plain("service", DataType::Utf8, false), + Field::plain("weight", DataType::Float64, false), + ])) +} + +fn batch(rows: &[(&str, &str, f64)]) -> Batch { + Batch::try_new( + input_schema(), + rows.iter() + .map(|(job, service, weight)| { + vec![ + Value::Utf8((*job).into()), + Value::Utf8((*service).into()), + Value::Float64(*weight), + ] + }) + .collect(), + ) + .unwrap() +} + +/// The planner contract: item = the counted column, weight = unit count or a +/// non-negative weight column, reduction = the group-by keys. +fn summary_agg(update: SummaryUpdate) -> PhysicalASAPDAGNode { + node( + 1, + Payload::SummaryAgg { + family: family(), + input: update, + reduction: Reduction::Reduce(GroupKeys::by(vec![0])), + grouping: grouping(), + filter: None, + }, + schema(vec![ + Field::plain("job", DataType::Utf8, false), + Field { + name: "state".into(), + dtype: family(), + nullable: false, + table: None, + }, + ]), + ) +} + +fn estimate(id: u32, query: SketchStatistic, dtype: DataType) -> PhysicalASAPDAGNode { + node( + id, + Payload::SummaryEstimate { query }, + schema(vec![ + Field::plain("job", DataType::Utf8, false), + Field::plain("value", dtype, false), + ]), + ) +} + +/// Compile `SummaryAgg` per input batch (merged when there are several), +/// then both readouts; return `(job → count, job → checkout frequency)`. +fn run( + update: SummaryUpdate, + batches: Vec, +) -> (BTreeMap, BTreeMap) { + let input = input_schema(); + let agg = summary_agg(update); + let build = compile_node(&agg, std::slice::from_ref(&input)).unwrap(); + let state = build.schema(); + let mut operators = BTreeMap::new(); + let mut contracts = BTreeMap::new(); + let mut sources = BTreeMap::new(); + let mut states = vec![]; + for (i, batch) in batches.into_iter().enumerate() { + let (source, built) = (i as u64, 100 + i as u64); + contracts.insert(source, InputContract::bounded(input.clone())); + sources.insert( + source, + Box::new(Operator::source(input.clone(), vec![batch]).unwrap()) as Source<'_>, + ); + operators.insert(built, (vec![source], build.clone())); + states.push(built); + } + let merged = if states.len() == 1 { + states[0] + } else { + let union = Operator::union(state.clone(), states.len()).unwrap(); + operators.insert(200, (states, union)); + let merge = compile_node( + &node(2, Payload::SummaryMerge, (*state).clone()), + std::slice::from_ref(&state), + ) + .unwrap(); + operators.insert(201, (vec![200], merge)); + 201 + }; + let count = estimate( + 3, + SketchStatistic::PointCount { + key: ColumnRef::SampleValue, + value: None, + }, + DataType::Int64, + ); + let point = estimate( + 4, + SketchStatistic::PointCount { + key: ColumnRef::Named("service".into()), + value: Some("checkout".into()), + }, + DataType::Float64, + ); + for (id, node) in [(300, &count), (301, &point)] { + let operator = compile_node(node, std::slice::from_ref(&state)).unwrap(); + operators.insert(id, (vec![merged], operator)); + } + let compiled = + CompiledPhysicalDAG::from_operators(contracts, operators, vec![300, 301]).unwrap(); + let dag = compiled.instantiate(sources).unwrap(); + let context = RunContext::new( + Scope::Query { + evaluation_time_ms: 0, + revision: 1, + }, + Limits::default(), + ) + .unwrap(); + let outputs = block_on(futures::future::join_all( + dag.execute(&[300, 301], context) + .unwrap() + .into_iter() + .map(|stream| stream.collect::>()), + )); + let mut outputs = outputs.into_iter().map(|stream| { + stream + .into_iter() + .flat_map(|batch| batch.unwrap().rows().to_vec()) + .collect::>() + }); + let rows = |rows: Vec>| { + rows.into_iter() + .map(|row| match &row[0] { + Value::Utf8(job) => (job.to_string(), row[1].clone()), + other => panic!("unexpected group {other:?}"), + }) + .collect::>() + }; + let counts = rows(outputs.next().unwrap()) + .into_iter() + .map(|(job, value)| match value { + Value::Int64(count) => (job, count), + other => panic!("count must be Int64, got {other:?}"), + }) + .collect(); + let points = rows(outputs.next().unwrap()) + .into_iter() + .map(|(job, value)| match value { + Value::Float64(point) => (job, point), + other => panic!("point must be Float64, got {other:?}"), + }) + .collect(); + (counts, points) +} + +fn exact(weighted: bool, job: &str, service: Option<&str>) -> f64 { + ROWS.iter() + .filter(|(j, s, _)| *j == job && service.is_none_or(|service| service == *s)) + .map(|(_, _, w)| if weighted { *w } else { 1. }) + .sum() +} + +/// Every estimate is at least the exact value and within the Count-Min +/// additive slack `e·N/w` of both Hydra levels (shared cell and inner sketch). +fn assert_within_cms_bound( + weighted: bool, + counts: &BTreeMap, + points: &BTreeMap, +) { + let total: f64 = ROWS + .iter() + .map(|(_, _, w)| if weighted { *w } else { 1. }) + .sum(); + let slack = + std::f64::consts::E * total * (1. / f64::from(SHARED_COLUMNS) + 1. / f64::from(WIDTH)); + assert_eq!(counts.len(), 3); + assert_eq!(points.len(), 3); + for job in ["api", "batch", "db"] { + let (count, true_count) = (counts[job] as f64, exact(weighted, job, None)); + assert!( + count >= true_count && count <= true_count + slack, + "{job}: {count}" + ); + let (point, true_point) = (points[job], exact(weighted, job, Some("checkout"))); + assert!( + point >= true_point && point <= true_point + slack, + "{job}: {point}" + ); + } +} + +// A unit-count HydraCms SummaryAgg compiles, runs, and answers per-group +// counts and per-group item frequencies within the CMS bound. +#[test] +fn unit_count_hydra_cms_answers_group_counts_and_points() { + let update = SummaryUpdate { + item: Some(SummaryInputExpr::Column(ColumnRef::Named("service".into()))), + weight: SummaryInputExpr::Constant(1.0), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount, + }, + }; + let (counts, points) = run(update, vec![batch(&ROWS)]); + assert_within_cms_bound(false, &counts, &points); +} + +// A non-negative weight column feeds HydraCms, and merging two builds over a +// split stream answers exactly as one build over the whole stream. +#[test] +fn weighted_hydra_cms_merges_like_one_build() { + let update = SummaryUpdate { + item: Some(SummaryInputExpr::Column(ColumnRef::Named("service".into()))), + weight: SummaryInputExpr::Column(ColumnRef::Named("weight".into())), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::CounterSamples, + }, + }; + let whole = run(update.clone(), vec![batch(&ROWS)]); + assert_within_cms_bound(true, &whole.0, &whole.1); + let (left, right) = ROWS.split_at(5); + assert_eq!(run(update, vec![batch(left), batch(right)]), whole); +} + +// HydraCms needs a non-negative weight proof; a signed weight is rejected at binding. +#[test] +fn signed_weight_is_rejected() { + let update = SummaryUpdate { + item: Some(SummaryInputExpr::Column(ColumnRef::Named("service".into()))), + weight: SummaryInputExpr::Column(ColumnRef::Named("weight".into())), + weight_domain: WeightDomain::UnknownOrSigned, + }; + assert!(compile_node(&summary_agg(update), &[input_schema()]).is_err()); +} diff --git a/crates/types/src/ir/schema/mod.rs b/crates/types/src/ir/schema/mod.rs index 67dbb8ec..b863303f 100644 --- a/crates/types/src/ir/schema/mod.rs +++ b/crates/types/src/ir/schema/mod.rs @@ -163,12 +163,19 @@ impl FieldDataType { /// state of their union with the family's guarantee intact. Rate/Increase /// accumulators depend on window edges, and merged heap top-k states have /// no accuracy model yet; families not listed fail closed. + /// + /// A shared Hydra grid over Count-Min or Count-Sketch cells is linear: + /// same-shape grids add cell-wise under the same hashes, giving the grid + /// of the union, so Theorem 2 still holds. HydraKll has no accuracy model. pub fn family_merges(&self) -> bool { - use state_type::{ExactKind as E, SketchAlgorithm as S}; + use state_type::{ExactKind as E, HydraKind as H, SketchAlgorithm as S}; match self { FieldDataType::ExactAggregate(kind, _) => { matches!(kind, E::Sum | E::Count | E::Min | E::Max) } + FieldDataType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind, .. }) => { + matches!(kind, H::HydraCms | H::HydraCountSketch) + } FieldDataType::Sketch(kind, _) => matches!( kind.algorithm(), S::Kll | S::DDSketch | S::Hll | S::Cms | S::CountSketch | S::UnivMon diff --git a/crates/types/tests/summary_merge_structure.rs b/crates/types/tests/summary_merge_structure.rs index 827fda0e..cde1b2e4 100644 --- a/crates/types/tests/summary_merge_structure.rs +++ b/crates/types/tests/summary_merge_structure.rs @@ -219,3 +219,41 @@ fn family_merge_capability() { assert_eq!(family.family_merges(), merges, "{family:?}"); } } + +/// Shared Hydra grids merge only for linear counter cells (#580 W7). +#[test] +fn hydra_merge_capability() { + use asap_types::ir::schema::{HydraKind, HydraParams}; + let hydra = |algorithm, params, kind, hydra_params| { + FieldDataType::Sketch( + SketchKind::new(algorithm, params), + GroupingStrategy::SharedMultiSubpopulation { + kind, + params: hydra_params, + }, + ) + }; + let (width, depth) = (64, 3); + let cms = hydra( + SketchAlgorithm::Cms, + SketchParams::Cms { width, depth }, + HydraKind::HydraCms, + HydraParams::HydraCms { + width, + depth, + shared_rows: 3, + shared_columns: 64, + }, + ); + assert!(cms.family_merges()); + let kll = hydra( + SketchAlgorithm::Kll, + SketchParams::Kll { k: 200 }, + HydraKind::HydraKll, + HydraParams::HydraKll { + k: 200, + shared_buckets: 200, + }, + ); + assert!(!kll.family_merges()); +}