diff --git a/crates/asap-physical-operators/src/capability.rs b/crates/asap-physical-operators/src/capability.rs index d48784548..5d5656d54 100644 --- a/crates/asap-physical-operators/src/capability.rs +++ b/crates/asap-physical-operators/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) => { @@ -116,6 +116,94 @@ pub fn validate_summary_kernel( Ok(()) } +/// The only shared grouping with a kernel: Hydra over Count-Min. Each update +/// adds one item column's value to its group, with weight 1 (a row count) or +/// a weight column proven non-negative (a Count-Min point count only +/// overestimates when no weight is negative). +fn validate_hydra_cms( + family: &SummaryFamilyType, + input: &SummaryUpdate, + grouping: &GroupingStrategy, +) -> Result<(), String> { + use planner_types::post_asap::{NonNegativeWeightProof, SummaryInputExpr, WeightDomain}; + if !matches!(family, SummaryFamilyType::Sketch(_, layout) if layout == grouping) { + return Err("Planner family and operator grouping disagree".into()); + } + hydra_cms_shape(family)?; + if !matches!(input.item, Some(SummaryInputExpr::Column(_))) { + return Err("HydraCms requires one item column".into()); + } + let WeightDomain::NonNegative { proof } = &input.weight_domain else { + return Err("HydraCms weights must be non-negative; the weight's sign is unknown".into()); + }; + match (&input.weight, proof) { + (SummaryInputExpr::Constant(w), NonNegativeWeightProof::UnitCount) if *w == 1.0 => Ok(()), + (SummaryInputExpr::Column(_), _) => Ok(()), + _ => Err("HydraCms weight must be the unit count or a non-negative weight column".into()), + } +} + +/// `(shared_rows, shared_columns, width, depth)` of a HydraCms family. The +/// 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, +) -> Result<(usize, usize, usize, usize), String> { + use planner_types::post_asap::{HydraKind, HydraParams}; + let SummaryFamilyType::Sketch( + kind, + GroupingStrategy::SharedMultiSubpopulation { + kind: HydraKind::HydraCms, + params: + HydraParams::HydraCms { + width, + depth, + shared_rows, + shared_columns, + }, + }, + ) = family + else { + return Err("shared summary grouping has no registered kernel".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, + )) +} + +/// A filtered `SummaryAgg` never runs at ingestion time. An ingestion-time +/// build reads raw samples (`$population`, `$timestamp`, `value`), which carry +/// none of the query's columns for the filter predicate to read, so only a +/// query-time build can apply it. The ingestion-time compiler and the +/// binding of any ingestion-timed `SummaryAgg` apply this check. Planner-side +/// timing validation (materialization timings, Stage 2 eligibility) does not +/// apply it yet: until it gains a check or a `PruneReason`, a plan placing a +/// filtered `SummaryAgg` at ingestion is accepted by the planner and rejected +/// here at compile time. +pub fn validate_ingestion_time_summary( + filter: Option<&planner_types::ir::Predicate>, +) -> Result<(), String> { + match filter { + Some(_) => Err("a filtered summary update runs only at query time".into()), + None => Ok(()), + } +} + fn valid_matrix(width: u32, depth: u32) -> bool { // Construction uses the kernel's native row hashing, so no encoded-size // limit applies here. @@ -147,9 +235,13 @@ pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error> { use planner_types::post_asap::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).map(|_| ()).map_err(Error::Invalid); + } + return if valid_matrix(*width, *depth) { Ok(()) } else { Err(Error::Invalid( @@ -212,6 +304,21 @@ pub fn validate_sketch_evaluation( (A::DDSketch, _) => bare_count, (A::Hll, SketchStatistic::Cardinality) => true, (A::Hll, _) => bare_count, + // HydraCms answers a group's item frequency. A group's total + // needs a group-mass query asap_sketchlib's Hydra lacks. + (A::Cms, SketchStatistic::PointCount { value, .. }) + if matches!(family, SummaryFamilyType::Sketch(_, grouping) + if grouping != &GroupingStrategy::PerSubpopulationInstance) => + { + if value.is_none() { + return Err(Error::Invalid( + "HydraCms does not answer a group total: asap_sketchlib's Hydra has no \ + group-mass query; only an item's point count within a group is answered" + .into(), + )); + } + 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, @@ -259,3 +366,18 @@ pub fn validate_exact_evaluation( } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + // Only a query-time build applies a summary filter. + #[test] + fn filtered_summary_runs_only_at_query_time() { + let filter = planner_types::ir::Predicate::(planner_types::ir::ScalarExpr::Literal( + planner_types::pre_asap::ScalarValue::Boolean(true), + )); + assert!(validate_ingestion_time_summary(Some(&filter)).is_err()); + assert!(validate_ingestion_time_summary::(None).is_ok()); + } +} diff --git a/crates/asap-physical-operators/src/operators/aggregate/mod.rs b/crates/asap-physical-operators/src/operators/aggregate/mod.rs index 7b51b3c66..0ef4c6f7d 100644 --- a/crates/asap-physical-operators/src/operators/aggregate/mod.rs +++ b/crates/asap-physical-operators/src/operators/aggregate/mod.rs @@ -48,11 +48,56 @@ impl Operator { kind: Kind::Aggregate { groups, measures: measures.into_iter().map(|(_, r)| r).collect(), + filters: vec![], }, inputs: vec![input], output: schema(fields), }) } + + /// Give each measure an optional row filter (SQL `FILTER (WHERE …)`). + /// A group is formed from all its rows, so a group with no matching row + /// still appears, with the measure's empty-input value: 0 for counts, + /// NULL for SQL SUM/AVG/MIN/MAX. No filter at all keeps the plain form. + pub fn with_measure_filters(mut self, filters: Vec>) -> Result { + let Kind::Aggregate { + filters: slot, + measures, + groups, + } = &mut self.kind + else { + return Err(invalid("measure filters need an aggregate")); + }; + if filters.iter().all(Option::is_none) { + return Ok(self); + } + if filters.len() != measures.len() { + return Err(invalid("one filter slot per aggregate measure required")); + } + // PromQL emits no per-measure filters (label matchers are scan + // predicates), and its outputs are non-nullable. + if self.inputs[0].has_promql_series_identity() { + return Err(invalid( + "measure filters are not supported over PromQL series identity", + )); + } + let mut output = (*self.output).clone(); + for (index, filter) in filters.iter().enumerate() { + let Some(filter) = filter else { continue }; + if filter.dtype(&self.inputs[0])?.0 != DataType::Bool { + return Err(invalid("measure filter must be boolean")); + } + if matches!( + measures[index], + Reduction::Sum(_) | Reduction::Avg(_) | Reduction::Min(_) | Reduction::Max(_) + ) { + output.fields[groups.len() + index].nullable = true; + } + } + *slot = filters; + self.output = Arc::new(output); + Ok(self) + } pub fn window( input: SchemaRef, intent: planner_types::pre_asap::AggIntent, @@ -163,8 +208,20 @@ pub(super) fn execute<'a>( ) .await? } - Kind::Aggregate { groups, measures } => { - reduce(rows, groups, measures, &operator.inputs[0], &context).await? + Kind::Aggregate { + groups, + measures, + filters, + } => { + reduce( + rows, + groups, + measures, + filters, + &operator.inputs[0], + &context, + ) + .await? } _ => unreachable!(), }; @@ -178,6 +235,7 @@ async fn reduce( rows: Vec>, groups: &[usize], measures: &[Reduction], + filters: &[Option], input: &SchemaRef, context: &RunContext, ) -> Result>, Error> { @@ -204,8 +262,21 @@ async fn reduce( .iter() .map(|&i| rows[0][i].clone()) .collect::>(); - for measure in measures { - result.push(reduce_one(&rows, measure, input, &mut work).await?); + for (index, measure) in measures.iter().enumerate() { + let mut selected = Vec::with_capacity(rows.len()); + for row in &rows { + // SQL FILTER keeps a row only when the predicate is true, not NULL. + if filters + .get(index) + .and_then(Option::as_ref) + .map_or(Ok(true), |filter| { + filter.evaluate(row).map(|v| matches!(v, Value::Bool(true))) + })? + { + selected.push(row); + } + } + result.push(reduce_one(&selected, measure, input, &mut work).await?); } workspace.grow(row_bytes(&result))?; output.push(result); @@ -239,7 +310,7 @@ pub(super) fn quantile(q: f64, mut values: Vec) -> f64 { } async fn reduce_one( - rows: &[Vec], + rows: &[&Vec], measure: &Reduction, input: &SchemaRef, work: &mut Cooperative, diff --git a/crates/asap-physical-operators/src/operators/mod.rs b/crates/asap-physical-operators/src/operators/mod.rs index 517019118..10cd1303f 100644 --- a/crates/asap-physical-operators/src/operators/mod.rs +++ b/crates/asap-physical-operators/src/operators/mod.rs @@ -123,6 +123,9 @@ enum Kind { Aggregate { groups: Vec, measures: Vec, + /// Per-measure row filters (SQL `FILTER (WHERE …)`); empty when none. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + filters: Vec>, }, SemiJoin { keys: Vec<(usize, usize)>, @@ -137,12 +140,30 @@ enum Kind { value: usize, time: Option, groups: Vec, + /// Rows for which this is not true update no state; their group is kept. + #[serde(default, skip_serializing_if = "Option::is_none")] + filter: Option>, }, KeyedSummaryBuild { family: SummaryFamilyType, value: usize, items: Vec, groups: Vec, + /// Rows for which this is not true update no state; their group is kept. + #[serde(default, skip_serializing_if = "Option::is_none")] + filter: Option>, + }, + /// One HydraCms state shared by all groups, adding each row's item weight. + SharedSummaryBuild { + family: SummaryFamilyType, + item: usize, + /// The weight column; each row adds 1 without one. + #[serde(default, skip_serializing_if = "Option::is_none")] + weight: Option, + groups: Vec, + /// Rows for which this is not true update no state; their group is kept. + #[serde(default, skip_serializing_if = "Option::is_none")] + filter: Option>, }, KeyedEvaluation { state: usize, @@ -287,6 +308,7 @@ impl PhysicalOperator for Operator { | Kind::SemiJoin { .. } | Kind::SummaryBuild { .. } | Kind::KeyedSummaryBuild { .. } + | Kind::SharedSummaryBuild { .. } | Kind::SummaryMerge { .. } | Kind::VectorToScalar { .. } ) @@ -337,7 +359,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", @@ -392,6 +416,7 @@ impl PhysicalOperator for Operator { Kind::SummaryBuild { .. } | Kind::Evaluation { .. } | Kind::KeyedSummaryBuild { .. } + | Kind::SharedSummaryBuild { .. } | Kind::KeyedEvaluation { .. } => summary::execute(self, inputs, context), } } diff --git a/crates/asap-physical-operators/src/operators/summary/mod.rs b/crates/asap-physical-operators/src/operators/summary/mod.rs index 46d82c6e5..a180948ac 100644 --- a/crates/asap-physical-operators/src/operators/summary/mod.rs +++ b/crates/asap-physical-operators/src/operators/summary/mod.rs @@ -56,11 +56,78 @@ impl Operator { value, items, groups, + filter: None, }, inputs: vec![input], output: schema(fields), }) } + /// One HydraCms state shared by every group, emitted as one row per group + /// that reads it. Each row adds its `weight` column's value (1 without + /// one) to its `item` value; a row whose weight is NULL adds nothing. + 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)?, + (DataType::Utf8 | DataType::Int64 | DataType::Bool, false) + ) { + return Err(invalid( + "HydraCms items must be non-null Utf8, Int64 or Bool", + )); + } + if let Some(weight) = weight { + if !matches!( + plain(&input, weight)?.0, + DataType::Float64 | DataType::Int64 + ) { + return Err(invalid("HydraCms weights must be Float64 or Int64")); + } + } + 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, + filter: None, + }, + inputs: vec![input], + output: schema(fields), + }) + } + /// Update the summary only from rows where `filter` is true (a filtered + /// `SummaryAgg`). Every group still gets a state, so a group with no + /// matching row reads as an empty summary. + pub fn with_row_filter(mut self, filter: Expression) -> Result { + if filter.dtype(&self.inputs[0])?.0 != DataType::Bool { + return Err(invalid("summary filter must be boolean")); + } + match &mut self.kind { + Kind::SummaryBuild { filter: slot, .. } + | Kind::KeyedSummaryBuild { filter: slot, .. } + | Kind::SharedSummaryBuild { filter: slot, .. } => *slot = Some(Box::new(filter)), + _ => return Err(invalid("a row filter needs a summary build")), + } + Ok(self) + } pub fn keyed_evaluation( input: SchemaRef, state: usize, @@ -142,6 +209,7 @@ impl Operator { value, time, groups, + filter: None, }, inputs: vec![input], output: schema(fields), @@ -240,10 +308,11 @@ pub(super) fn execute<'a>( value, time, groups, + filter, } => Ok(futures::stream::once(async move { Batch::try_new( output, - build_summary(input, family, *value, *time, groups, !operator.inputs[0].has_promql_series_identity(), &context).await?, + build_summary(input, family, *value, *time, groups, filter.as_deref(), !operator.inputs[0].has_promql_series_identity(), &context).await?, ) }) .boxed_local()), @@ -252,10 +321,26 @@ pub(super) fn execute<'a>( value, items, groups, + filter, } => Ok(futures::stream::once(async move { Batch::try_new( output, - build_keyed_summary(input, family, *value, items, groups, &context).await?, + build_keyed_summary(input, family, *value, items, groups, filter.as_deref(), &context) + .await?, + ) + }) + .boxed_local()), + Kind::SharedSummaryBuild { + family, + item, + weight, + groups, + filter, + } => Ok(futures::stream::once(async move { + Batch::try_new( + output, + build_shared_summary(input, family, *item, *weight, groups, filter.as_deref(), &context) + .await?, ) }) .boxed_local()), @@ -302,6 +387,12 @@ pub(super) fn execute<'a>( return Err(invalid("summary value required")); }; row[*state] = match query { + // A filtered or NULL-only group's quantile is SQL NULL. + SummaryEvaluation::Sketch(_) + if output.fields[*state].nullable && summary.is_empty() => + { + Value::Null + } SummaryEvaluation::Sketch(query) => { let value = summary .estimate(query) @@ -376,12 +467,14 @@ pub(super) fn execute_merge<'a>( .boxed_local()) } +#[allow(clippy::too_many_arguments)] async fn build_summary( mut input: Input<'_, Batch>, family: &SummaryFamilyType, value: usize, time: Option, groups: &[usize], + filter: Option<&Expression>, emit_empty_global: bool, context: &RunContext, ) -> Result>, Error> { @@ -432,6 +525,9 @@ async fn build_summary( )?; states.insert(key.clone(), state); } + if !selected(filter, row)? { + continue; + } let (_, updater, memory, overhead, previous) = states.get_mut(&key).expect("inserted group"); // SQL aggregates ignore NULL samples while retaining the group. @@ -474,12 +570,20 @@ async fn build_summary( }) .collect()) } +/// Whether a row passes a summary build's filter: only a true predicate +/// does, as with SQL `FILTER (WHERE …)`. +fn selected(filter: Option<&Expression>, row: &[Value]) -> Result { + filter.map_or(Ok(true), |filter| { + Ok(matches!(filter.evaluate(row)?, Value::Bool(true))) + }) +} async fn merge_summary( mut rows: Vec>, state_column: usize, groups: &[usize], context: &RunContext, ) -> Result>, Error> { + use crate::summary_kernels::{HydraCms, HydraCmsGroup}; // Counter states join consecutive windows pairwise, which is exact only // in time order: order the inputs by their window start. if let Some(Value::Summary { family, .. }) = rows.first().map(|row| &row[state_column]) { @@ -500,16 +604,64 @@ async fn merge_summary( let mut work = Cooperative::new(context); let mut memory = context.reserve(0)?; let mut retained = 0usize; + // States shared by several groups (Hydra grids) are reserved once each. + let mut shared = BTreeMap::::new(); + // Each distinct pair of Hydra grids merges once, and every group of the + // pair reads the one merged grid. The inputs are kept so their addresses + // stay unique while they key the memo. + type GridPair = (Arc, Arc, Arc); + let mut grids = BTreeMap::<(usize, usize), GridPair>::new(); for row in rows { work.checkpoint().await?; let Value::Summary { family, state } = &row[state_column] else { return Err(invalid("summary state required")); }; + if let Some((id, bytes)) = state.shared_state() { + if shared.insert(id, bytes).is_none() { + retained = retained.checked_add(bytes).ok_or(Error::MemoryLimit)?; + memory.resize(retained)?; + } + } let key = group_key(&row, groups)?; if let Some((_, expected, existing)) = states.get_mut(&key) { if expected != family { return Err(invalid("incompatible summary family")); } + if let (Some(left), Some(right)) = ( + existing.as_any().downcast_ref::(), + state.as_any().downcast_ref::(), + ) { + if left.group != right.group { + return Err(invalid("HydraCms merge requires the same group")); + } + let pair = ( + Arc::as_ptr(&left.grid) as usize, + Arc::as_ptr(&right.grid) as usize, + ); + let grid = match grids.entry(pair) { + std::collections::btree_map::Entry::Occupied(entry) => entry.get().2.clone(), + std::collections::btree_map::Entry::Vacant(entry) => { + let merged = Arc::new( + left.grid + .merge(&right.grid) + .map_err(|e| Error::Operator(e.to_string()))?, + ); + let bytes = merged.approx_memory_bytes(); + shared.insert(Arc::as_ptr(&merged) as usize, bytes); + retained = retained.checked_add(bytes).ok_or(Error::MemoryLimit)?; + memory.resize(retained)?; + entry + .insert((left.grid.clone(), right.grid.clone(), merged)) + .2 + .clone() + } + }; + *existing = Arc::new(HydraCmsGroup { + grid, + group: left.group.clone(), + }); + continue; + } let old_bytes = existing.approx_memory_bytes(); // Reserve an estimate for the replacement while both input states remain live. memory.resize( @@ -558,6 +710,7 @@ async fn build_keyed_summary( value: usize, items: &[usize], groups: &[usize], + filter: Option<&Expression>, context: &RunContext, ) -> Result>, Error> { use crate::{summary_kernels::weighted_frequency::WeightedFrequency, AggregateCore}; @@ -594,6 +747,9 @@ async fn build_keyed_summary( ), ); } + if !selected(filter, row)? { + continue; + } let (_, summary, reservation, overhead) = states.get_mut(&key).unwrap(); let Value::Float64(weight) = row[value] else { return Err(invalid("weighted frequency weight type")); @@ -622,3 +778,93 @@ async fn build_keyed_summary( }) .collect()) } + +async fn build_shared_summary( + mut input: Input<'_, Batch>, + family: &SummaryFamilyType, + item: usize, + weight: Option, + groups: &[usize], + filter: Option<&Expression>, + 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)); + } + if !selected(filter, row)? { + continue; + } + let weight = match weight.map(|i| &row[i]) { + None => 1, + Some(Value::Null) => continue, + Some(value) => hydra_weight(value)?, + }; + 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.count(&seen[&key].1, &item, weight) + .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()) +} + +/// One row's HydraCms weight. sketchlib's Hydra adds `i32` weights, so a +/// weight must be a non-negative integer within `i32`. +fn hydra_weight(value: &Value) -> Result { + let weight = match value { + Value::Int64(v) => *v as f64, + Value::Float64(v) => *v, + _ => return Err(Error::Operator("HydraCms weight type".into())), + }; + if weight < 0.0 { + return Err(Error::Operator(format!( + "HydraCms weights must be non-negative, got {weight}" + ))); + } + if !(weight.is_finite() && weight.fract() == 0.0 && weight <= f64::from(i32::MAX)) { + return Err(Error::Operator(format!( + "HydraCms weights must be integers within i32 (asap_sketchlib's Hydra adds i32 weights), got {weight}" + ))); + } + Ok(weight as i32) +} diff --git a/crates/asap-physical-operators/src/operators/unchecked.rs b/crates/asap-physical-operators/src/operators/unchecked.rs index 1b5dd7c4f..9f6bd0799 100644 --- a/crates/asap-physical-operators/src/operators/unchecked.rs +++ b/crates/asap-physical-operators/src/operators/unchecked.rs @@ -116,12 +116,17 @@ impl TryFrom for Operator { groups, window, } => Operator::window(input(0)?, *intent, coordinate, value, groups, window)?, - Kind::Aggregate { groups, measures } => { + Kind::Aggregate { + groups, + measures, + filters, + } => { if groups.len() + measures.len() != output.fields.len() { return Err(invalid("aggregate width mismatch")); } let names = output.fields[groups.len()..].iter().map(|f| f.name.clone()); Operator::aggregate(input(0)?, groups, names.zip(measures).collect())? + .with_measure_filters(filters)? } Kind::SemiJoin { keys, @@ -146,13 +151,31 @@ impl TryFrom for Operator { value, time, groups, - } => Operator::summary_build(input(0)?, family, value, time, groups)?, + filter, + } => with_filter( + Operator::summary_build(input(0)?, family, value, time, groups)?, + filter, + )?, Kind::KeyedSummaryBuild { family, value, items, groups, - } => Operator::keyed_summary_build(input(0)?, family, value, items, groups)?, + filter, + } => with_filter( + Operator::keyed_summary_build(input(0)?, family, value, items, groups)?, + filter, + )?, + Kind::SharedSummaryBuild { + family, + item, + weight, + groups, + filter, + } => with_filter( + Operator::shared_summary_build(input(0)?, family, item, weight, groups)?, + filter, + )?, Kind::KeyedEvaluation { state, k } => { Operator::keyed_evaluation(input(0)?, state, k, output.clone())? } @@ -173,3 +196,10 @@ impl TryFrom for Operator { Ok(op) } } + +fn with_filter(operator: Operator, filter: Option>) -> Result { + match filter { + Some(filter) => operator.with_row_filter(*filter), + None => Ok(operator), + } +} diff --git a/crates/asap-physical-operators/src/physical_planner/mod.rs b/crates/asap-physical-operators/src/physical_planner/mod.rs index 19e869417..36fc2a247 100644 --- a/crates/asap-physical-operators/src/physical_planner/mod.rs +++ b/crates/asap-physical-operators/src/physical_planner/mod.rs @@ -748,9 +748,6 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result { - if filters.iter().any(Option::is_some) { - return Err(invalid("filtered aggregate has no native implementation")); - } if measures.len() != output_names.len() { return Err(invalid("aggregate output names differ from measures")); } @@ -782,7 +779,13 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result>()?; - Operator::aggregate(input.clone(), groups(input, keys)?, measures) + Operator::aggregate(input.clone(), groups(input, keys)?, measures)? + .with_measure_filters( + filters + .iter() + .map(|f| f.as_ref().map(|f| expression(&f.0, input)).transpose()) + .collect::>()?, + ) } _ => Err(invalid("value operation has no native implementation")), }, @@ -794,75 +797,16 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result { - if filter.is_some() { - return Err(invalid( - "filtered summary update has no native implementation", - )); + if node.output_state.timing == planner_types::post_asap::ExecutionTiming::IngestionTime + { + crate::capability::validate_ingestion_time_summary(filter.as_ref()) + .map_err(Error::Invalid)?; } - if let Some(item) = &update.item { - let PlannerReduction::Reduce(keys) = reduction else { - return Err(invalid("keyed summary requires explicit partitions")); - }; - let SummaryInputExpr::Column(weight) = &update.weight else { - return Err(invalid( - "keyed summary weight must be a finalized value column", - )); - }; - if matches!(family, FieldDataType::Sketch(kind, _) if kind.algorithm() == &planner_types::post_asap::SketchAlgorithm::CmsWithHeap) - && !matches!( - update.weight_domain, - planner_types::post_asap::WeightDomain::NonNegative { .. } - ) - { - return Err(invalid("CMS requires a nonnegative weight contract")); - } - fn columns( - expr: &SummaryInputExpr, - input: &SchemaRef, - result: &mut Vec, - ) -> Result<(), Error> { - match expr { - SummaryInputExpr::Column(column) => { - result.push(named_column(input, column)?) - } - SummaryInputExpr::Tuple(items) => { - for item in items { - columns(item, input, result)?; - } - } - _ => return Err(invalid("keyed summary needs explicit item columns")), - } - Ok(()) - } - let mut items = Vec::new(); - columns(item, input, &mut items)?; - return Operator::keyed_summary_build( - input.clone(), - family.clone(), - named_column(input, weight)?, - items, - groups(input, keys)?, - ); + let build = summary_build(input, family, update, reduction, grouping)?; + match filter { + Some(filter) => build.with_row_filter(expression(&filter.0, input)?), + None => Ok(build), } - crate::capability::validate_summary_kernel(family, update, grouping) - .map_err(Error::Invalid)?; - let SummaryInputExpr::Column(column) = &update.weight else { - return Err(invalid( - "summary update expression must be projected to a column", - )); - }; - let PlannerReduction::Reduce(keys) = reduction else { - return Err(invalid( - "summary construction requires explicit grouping columns", - )); - }; - Operator::summary_build( - input.clone(), - family.clone(), - named_column(input, column)?, - input.time_index, - groups(input, keys)?, - ) } Payload::ASAP(ASAPOp::SummaryMerge { .. }) => { let state = summary_column(input)?; @@ -894,6 +838,99 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result Result { + if grouping != &planner_types::post_asap::GroupingStrategy::PerSubpopulationInstance { + crate::capability::validate_summary_kernel(family, update, grouping) + .map_err(Error::Invalid)?; + let PlannerReduction::Reduce(keys) = reduction else { + return Err(invalid("shared summary requires explicit groups")); + }; + let Some(SummaryInputExpr::Column(item)) = &update.item else { + unreachable!("validated HydraCms item column") + }; + // Validated: the unit count, or a non-negative weight column. + let weight = match &update.weight { + SummaryInputExpr::Column(weight) => 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")); + }; + let SummaryInputExpr::Column(weight) = &update.weight else { + return Err(invalid( + "keyed summary weight must be a finalized value column", + )); + }; + if matches!(family, FieldDataType::Sketch(kind, _) if kind.algorithm() == &planner_types::post_asap::SketchAlgorithm::CmsWithHeap) + && !matches!( + update.weight_domain, + planner_types::post_asap::WeightDomain::NonNegative { .. } + ) + { + return Err(invalid("CMS requires a nonnegative weight contract")); + } + fn columns( + expr: &SummaryInputExpr, + input: &SchemaRef, + result: &mut Vec, + ) -> Result<(), Error> { + match expr { + SummaryInputExpr::Column(column) => result.push(named_column(input, column)?), + SummaryInputExpr::Tuple(items) => { + for item in items { + columns(item, input, result)?; + } + } + _ => return Err(invalid("keyed summary needs explicit item columns")), + } + Ok(()) + } + let mut items = Vec::new(); + columns(item, input, &mut items)?; + return Operator::keyed_summary_build( + input.clone(), + family.clone(), + named_column(input, weight)?, + items, + groups(input, keys)?, + ); + } + crate::capability::validate_summary_kernel(family, update, grouping).map_err(Error::Invalid)?; + let SummaryInputExpr::Column(column) = &update.weight else { + return Err(invalid( + "summary update expression must be projected to a column", + )); + }; + let PlannerReduction::Reduce(keys) = reduction else { + return Err(invalid( + "summary construction requires explicit grouping columns", + )); + }; + Operator::summary_build( + input.clone(), + family.clone(), + named_column(input, column)?, + input.time_index, + groups(input, keys)?, + ) +} fn summary_column(input: &SchemaRef) -> Result { let columns = input .fields diff --git a/crates/asap-physical-operators/src/physical_planner/precompute.rs b/crates/asap-physical-operators/src/physical_planner/precompute.rs index 672ea9750..ceaacdeef 100644 --- a/crates/asap-physical-operators/src/physical_planner/precompute.rs +++ b/crates/asap-physical-operators/src/physical_planner/precompute.rs @@ -413,11 +413,8 @@ fn fragment( filter, .. }) => { - if filter.is_some() { - return Err(invalid( - "filtered summary update has no native implementation", - )); - } + crate::capability::validate_ingestion_time_summary(filter.as_ref()) + .map_err(Error::Invalid)?; let [input] = schemas else { return Err(invalid("summary update requires one input")); }; diff --git a/crates/asap-physical-operators/src/summary_kernels/datasketches_kll.rs b/crates/asap-physical-operators/src/summary_kernels/datasketches_kll.rs index e4ecb56a0..14ed5a9ad 100644 --- a/crates/asap-physical-operators/src/summary_kernels/datasketches_kll.rs +++ b/crates/asap-physical-operators/src/summary_kernels/datasketches_kll.rs @@ -62,12 +62,19 @@ impl AggregateCore for DatasketchesKLLAccumulator { fn estimate(&self, query: &SketchStatistic) -> Result { match query { + SketchStatistic::Quantile { .. } if self.is_empty() => { + Err("KLL quantile of an empty population".into()) + } SketchStatistic::Quantile { q } if (0.0..=1.0).contains(q) => Ok(self.get_quantile(*q)), SketchStatistic::Quantile { .. } => Err("quantile must be in [0, 1]".into()), other => Err(format!("KLL does not answer {other:?}").into()), } } + fn is_empty(&self) -> bool { + self.inner.count() == 0 + } + fn approx_memory_bytes(&self) -> usize { // KLL with default k=200 holds ~2*k items (~3 KiB); round up for overhead. 4 * 1024 diff --git a/crates/asap-physical-operators/src/summary_kernels/dd_sketch.rs b/crates/asap-physical-operators/src/summary_kernels/dd_sketch.rs index e9cdf9d9c..d2f21cf29 100644 --- a/crates/asap-physical-operators/src/summary_kernels/dd_sketch.rs +++ b/crates/asap-physical-operators/src/summary_kernels/dd_sketch.rs @@ -38,6 +38,10 @@ impl AggregateCore for DDSketchAccumulator { })) } + fn is_empty(&self) -> bool { + self.inner.total_count() == 0 + } + /// Quantiles, and the total sample count as a bare `PointCount`. fn estimate(&self, query: &SketchStatistic) -> Result { match query { diff --git a/crates/asap-physical-operators/src/summary_kernels/factory.rs b/crates/asap-physical-operators/src/summary_kernels/factory.rs index ce6bd5c7a..9dba5d6ac 100644 --- a/crates/asap-physical-operators/src/summary_kernels/factory.rs +++ b/crates/asap-physical-operators/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}; @@ -546,7 +546,9 @@ pub fn create_planner_accumulator( crate::capability::validate_summary_kernel(family, input, grouping)?; use planner_types::post_asap::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 { @@ -636,6 +638,14 @@ 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 (shared_rows, shared_columns, width, depth) = crate::capability::hydra_cms_shape(family)?; + HydraCms::new(shared_rows, shared_columns, width, depth).map_err(|e| e.to_string()) +} + struct PlannerExactUpdater { acc: crate::summary_kernels::exact::ExactAccumulator, } @@ -802,4 +812,53 @@ 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::post_asap::{ + GroupingStrategy, HydraKind, HydraParams, NonNegativeWeightProof, WeightDomain, + }; + 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::pre_asap::ColumnRef::Named("host".into()), + )), + weight: SummaryInputExpr::Constant(1.0), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount, + }, + }; + assert!( + crate::capability::validate_summary_kernel(&family(128), &update, &grouping(128)) + .is_ok() + ); + assert!(create_planner_accumulator(&family(128), &update, &grouping(128)).is_err()); + } } diff --git a/crates/asap-physical-operators/src/summary_kernels/hydra_cms.rs b/crates/asap-physical-operators/src/summary_kernels/hydra_cms.rs new file mode 100644 index 000000000..e3393bab9 --- /dev/null +++ b/crates/asap-physical-operators/src/summary_kernels/hydra_cms.rs @@ -0,0 +1,273 @@ +//! 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 +//! adds up the group's item weights (1 per row for a count). Colliding groups share a cell, so a point count +//! only overestimates. +//! +//! The kernel answers a group's item frequency (Hydra's median estimator). +//! A group's total needs a group-mass query that `asap_sketchlib`'s Hydra +//! does not offer, so it is not answered here. +use crate::{AggregateCore, KernelError}; +use asap_sketchlib::{input::HydraCounter, CountMin, DataInput, FastPath, Hydra, Vector2D}; +use planner_types::post_asap::SketchStatistic; +use std::sync::Arc; + +/// The single Hydra key column: a group's whole label tuple is one +/// subpopulation, so no row fans out to 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 weight added. 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` in `group` (1 for a row count). + /// sketchlib's Hydra adds `i32` weights, so weights are non-negative + /// integers and their total stays within `i32`. + pub fn count(&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 count exceeds its i32 counters".into()); + } + 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))?) + } + + /// 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 cell 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 count 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) + } +} + +/// 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 valued point count reads that item's frequency within the group. + fn estimate(&self, query: &SketchStatistic) -> Result { + match query { + SketchStatistic::PointCount { + value: Some(item), .. + } => self.grid.point(&self.group, item), + _ => Err(format!("{query:?} is not supported by HydraCms").into()), + } + } + + /// The group's own view; the grid it shares is [`Self::shared_state`]. + fn approx_memory_bytes(&self) -> usize { + std::mem::size_of::() + self.group.len() + } + + fn shared_state(&self) -> Option<(usize, usize)> { + Some(( + Arc::as_ptr(&self.grid) as usize, + self.grid.approx_memory_bytes(), + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const ROWS: [(&str, &str); 9] = [ + ("api", "checkout"), + ("api", "checkout"), + ("api", "search"), + ("api", "a:b;c"), + ("batch", "checkout"), + ("batch", "export"), + ("db;x:y", "checkout"), + ("db;x:y", "vacuum"), + ("db;x:y", "vacuum"), + ]; + + fn exact(group: &str, item: &str) -> f64 { + ROWS.iter().filter(|r| **r == (group, item)).count() as f64 + } + + fn build(rows: &[(&str, &str)], shared_columns: usize) -> HydraCms { + let mut grid = HydraCms::new(3, shared_columns, 64, 3).unwrap(); + for (group, item) in rows { + grid.count(group, item, 1).unwrap(); + } + grid + } + + // Point counts stay within the CMS bound of exact values; N = 9, so the + // additive slack is e·N/64 per level. + #[test] + fn point_estimates_within_cms_bound() { + let grid = build(&ROWS, 64); + let slack = std::f64::consts::E * ROWS.len() as f64 / 64.0 * 2.0; + for group in ["api", "batch", "db;x:y"] { + for item in ["checkout", "search", "export", "vacuum", "a:b;c"] { + let estimate = grid.point(group, item).unwrap(); + let truth = exact(group, item); + assert!( + estimate >= truth && estimate <= truth + slack, + "{group}/{item}: {estimate} vs {truth}" + ); + } + } + } + + // With one shared column every group collides, so a point count reads + // the item's count over the whole stream: never below the exact value. + #[test] + fn collisions_only_overestimate() { + let grid = build(&ROWS, 1); + assert_eq!(grid.point("batch", "checkout").unwrap(), 4.0); + } + + // 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"] { + for item in ["checkout", "vacuum"] { + assert_eq!( + merged.point(group, item).unwrap(), + whole.point(group, item).unwrap() + ); + } + } + assert!(build(left, 8).merge(&build(right, 16)).is_err()); + } + + // Group views answer only their own group's item counts and merge only + // with the same group; a group total is not answered. + #[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 checkout = SketchStatistic::PointCount { + key: planner_types::pre_asap::ColumnRef::Named("item".into()), + value: Some("checkout".into()), + }; + assert_eq!(view("batch").estimate(&checkout).unwrap(), 1.0); + let merged = view("batch").merge_with(&view("batch")).unwrap(); + assert_eq!(merged.estimate(&checkout).unwrap(), 2.0); + assert!(view("api").merge_with(&view("batch")).is_err()); + let total = SketchStatistic::PointCount { + key: planner_types::pre_asap::ColumnRef::SampleValue, + value: None, + }; + assert!(view("api").estimate(&total).is_err()); + assert!(view("api").estimate(&SketchStatistic::Cardinality).is_err()); + } + + // Counts that could overflow the i32 counters are rejected. + #[test] + fn rejects_overflowing_counts() { + let mut grid = HydraCms::new(1, 1, 1, 1).unwrap(); + grid.total = i64::from(i32::MAX); + assert!(grid.count("g", "x", 1).is_err()); + assert!(grid.merge(&HydraCms::new(1, 1, 1, 1).unwrap()).is_ok()); + assert!(grid.merge(&grid).is_err()); + } +} diff --git a/crates/asap-physical-operators/src/summary_kernels/mod.rs b/crates/asap-physical-operators/src/summary_kernels/mod.rs index 478e75d09..095f82f27 100644 --- a/crates/asap-physical-operators/src/summary_kernels/mod.rs +++ b/crates/asap-physical-operators/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/asap-physical-operators/src/summary_kernels/traits.rs b/crates/asap-physical-operators/src/summary_kernels/traits.rs index 46e2c7f86..3ca12f4b0 100644 --- a/crates/asap-physical-operators/src/summary_kernels/traits.rs +++ b/crates/asap-physical-operators/src/summary_kernels/traits.rs @@ -26,10 +26,26 @@ pub trait AggregateCore: Send + Sync { Err(format!("{query:?} is not supported by this summary").into()) } + /// True when no sample reached this state, so a value readout (a + /// quantile) has no answer, which SQL reads as NULL. Kernels that do not + /// track it report false. + fn is_empty(&self) -> bool { + false + } + /// Approximate in-memory footprint, used for execution memory reservations. + /// A state shared with other rows (see [`Self::shared_state`]) is not + /// included. fn approx_memory_bytes(&self) -> usize { 4096 } + + /// The identity and footprint of a state this row shares with other + /// rows (one Hydra grid read by every group of a build). Reservations + /// count it once per identity, not once per row. + fn shared_state(&self) -> Option<(usize, usize)> { + None + } } impl Clone for Box { diff --git a/crates/asap-physical-operators/src/values.rs b/crates/asap-physical-operators/src/values.rs index 9b22f0a2e..598ffe883 100644 --- a/crates/asap-physical-operators/src/values.rs +++ b/crates/asap-physical-operators/src/values.rs @@ -239,6 +239,16 @@ impl Batch { &self.rows } pub fn bytes(&self) -> usize { + // A summary state shared by several rows is counted once. + let shared = self + .rows + .iter() + .flat_map(|r| r.iter()) + .filter_map(|v| match v { + Value::Summary { state, .. } => state.shared_state(), + _ => None, + }) + .collect::>(); std::mem::size_of::() + self.rows.capacity() * std::mem::size_of::>() + self @@ -247,6 +257,7 @@ impl Batch { .flat_map(|r| r.iter()) .map(Value::bytes) .sum::() + + shared.values().sum::() } } pub(crate) fn group_key(row: &[Value], columns: &[usize]) -> Result>, Error> { @@ -285,7 +296,16 @@ fn validate_state(family: &SummaryFamilyType, state: &dyn AggregateCore) -> Resu state.algorithm() == algorithm && state.shape() == (width, depth, capacity) }) } - + SummaryFamilyType::Sketch(_, grouping) + if grouping + != &planner_types::post_asap::GroupingStrategy::PerSubpopulationInstance => + { + let shape = crate::capability::hydra_cms_shape(family).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/asap-physical-operators/tests/hydra_cms_execution.rs b/crates/asap-physical-operators/tests/hydra_cms_execution.rs new file mode 100644 index 000000000..4fcb43615 --- /dev/null +++ b/crates/asap-physical-operators/tests/hydra_cms_execution.rs @@ -0,0 +1,417 @@ +//! A Hydra-grouped Count-Min `SummaryAgg` compiles through the physical +//! planner and answers each group's item point counts from one shared grid. +use asap_physical_operators::{ + 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::physical_export::{ + PhysicalASAPDAGNode, PhysicalASAPOperatorPayload as Payload, +}; +use planner_types::ir::{ASAPOp, ExprSemantics, Materialization, Predicate, ScalarExpr}; +use planner_types::post_asap::*; +use planner_types::pre_asap::{ + ColumnRef, CompareOpKind, DataType, GroupKeys, Reduction, ScalarValue, +}; +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: usize, payload: Payload, output: Schema) -> PhysicalASAPDAGNode { + PhysicalASAPDAGNode { + id, + payload, + output_state: ExecutionDataState::QUERY_ROWS, + materialization: Materialization::NotMaterialized, + output_schema: output, + guarantee: 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() +} + +fn unit_count() -> SummaryUpdate { + SummaryUpdate { + item: Some(SummaryInputExpr::Column(ColumnRef::Named("service".into()))), + weight: SummaryInputExpr::Constant(1.0), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount, + }, + } +} + +/// The planner contract: item = the counted column, weight = unit count, +/// reduction = the group-by keys. +fn summary_agg(update: SummaryUpdate, filter: Option>) -> PhysicalASAPDAGNode { + node( + 1, + Payload::ASAP(ASAPOp::SummaryAgg { + child: 0, + family: family(), + input: update, + reduction: Reduction::Reduce(GroupKeys::by(vec![0])), + grouping: grouping(), + filter, + }), + schema(vec![ + Field::plain("job", DataType::Utf8, false), + Field { + name: "state".into(), + dtype: family(), + nullable: false, + table: None, + }, + ]), + ) +} + +fn estimate(id: usize, query: SketchStatistic, dtype: DataType) -> PhysicalASAPDAGNode { + node( + id, + Payload::ASAP(ASAPOp::SummaryEstimate { + summary_input: 1, + query, + }), + schema(vec![ + Field::plain("job", DataType::Utf8, false), + Field::plain("value", dtype, false), + ]), + ) +} + +fn point(item: &str) -> SketchStatistic { + SketchStatistic::PointCount { + key: ColumnRef::Named("service".into()), + value: Some(item.into()), + } +} + +/// Compile `SummaryAgg` per input batch (merged when there are several), +/// then read each job's `checkout` frequency. +fn run( + update: SummaryUpdate, + filter: Option>, + batches: Vec, +) -> BTreeMap { + try_run(update, filter, batches).unwrap() +} + +/// As [`run`], returning the first execution error. +fn try_run( + update: SummaryUpdate, + filter: Option>, + batches: Vec, +) -> Result, String> { + let input = input_schema(); + let build = compile_node(&summary_agg(update, filter), 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::ASAP(ASAPOp::SummaryMerge { children: vec![1] }), + (*state).clone(), + ), + std::slice::from_ref(&state), + ) + .unwrap(); + operators.insert(201, (vec![200], merge)); + 201 + }; + let read = compile_node( + &estimate(3, point("checkout"), DataType::Float64), + std::slice::from_ref(&state), + ) + .unwrap(); + operators.insert(300, (vec![merged], read)); + let compiled = CompiledPhysicalDAG::from_operators(contracts, operators, vec![300]).unwrap(); + let dag = compiled.instantiate(sources).unwrap(); + let context = RunContext::new( + Scope::Query { + evaluation_time_ms: 0, + revision: 1, + }, + Limits::default(), + ) + .unwrap(); + let batches = block_on( + dag.execute(&[300], context) + .unwrap() + .remove(0) + .collect::>(), + ); + let mut points = BTreeMap::new(); + for batch in batches { + for row in batch.map_err(|e| e.to_string())?.rows() { + match (&row[0], &row[1]) { + (Value::Utf8(job), Value::Float64(point)) => { + points.insert(job.to_string(), *point); + } + other => panic!("unexpected row {other:?}"), + } + } + } + Ok(points) +} + +fn exact(job: &str, service: &str) -> f64 { + ROWS.iter() + .filter(|(j, s, _)| *j == job && *s == service) + .count() as f64 +} + +/// 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(points: &BTreeMap) { + let total = ROWS.len() as f64; + let slack = + std::f64::consts::E * total * (1. / f64::from(SHARED_COLUMNS) + 1. / f64::from(WIDTH)); + assert_eq!(points.len(), 3); + for job in ["api", "batch", "db"] { + let (point, truth) = (points[job], exact(job, "checkout")); + assert!(point >= truth && point <= truth + slack, "{job}: {point}"); + } +} + +// A unit-count HydraCms SummaryAgg compiles, runs, and answers per-group item +// frequencies within the CMS bound. +#[test] +fn unit_count_hydra_cms_answers_group_item_counts() { + assert_within_cms_bound(&run(unit_count(), None, vec![batch(&ROWS)])); +} + +// Merging two builds over a split stream answers exactly as one build over +// the whole stream. +#[test] +fn hydra_cms_merges_like_one_build() { + let whole = run(unit_count(), None, vec![batch(&ROWS)]); + let (left, right) = ROWS.split_at(5); + assert_eq!( + run(unit_count(), None, vec![batch(left), batch(right)]), + whole + ); +} + +fn weighted(weight_domain: WeightDomain) -> SummaryUpdate { + SummaryUpdate { + weight: SummaryInputExpr::Column(ColumnRef::Named("weight".into())), + weight_domain, + ..unit_count() + } +} + +fn non_negative() -> WeightDomain { + WeightDomain::NonNegative { + proof: NonNegativeWeightProof::ResetAwareCounterDerivative, + } +} + +// A non-negative weight column sums each item's weight within its group +// (e.g. bytes per service per job), within the Count-Min bound over the +// stream's total weight; a split-stream merge answers as one build. +#[test] +fn weighted_hydra_cms_sums_item_weights_per_group() { + let points = run(weighted(non_negative()), None, vec![batch(&ROWS)]); + let total: f64 = ROWS.iter().map(|(_, _, w)| w).sum(); + let slack = + std::f64::consts::E * total * (1. / f64::from(SHARED_COLUMNS) + 1. / f64::from(WIDTH)); + for job in ["api", "batch", "db"] { + let truth: f64 = ROWS + .iter() + .filter(|(j, s, _)| *j == job && *s == "checkout") + .map(|(_, _, w)| w) + .sum(); + let point = points[job]; + assert!(point >= truth && point <= truth + slack, "{job}: {point}"); + } + let (left, right) = ROWS.split_at(5); + assert_eq!( + run( + weighted(non_negative()), + None, + vec![batch(left), batch(right)] + ), + points + ); +} + +// A weight column of unknown sign is rejected at binding: a Count-Min point +// count only overestimates when no weight is negative. +#[test] +fn unknown_sign_weight_is_rejected() { + let error = compile_node( + &summary_agg(weighted(WeightDomain::UnknownOrSigned), None), + &[input_schema()], + ) + .err() + .expect("unknown-sign weight"); + assert!(error.to_string().contains("non-negative"), "{error}"); +} + +// sketchlib's Hydra adds integer weights, so a negative, fractional or +// non-finite weight value fails the build with a clear error. +#[test] +fn invalid_weight_values_fail_the_build() { + for (weight, expected) in [ + (-1.0, "non-negative"), + (1.5, "integer"), + (f64::NAN, "integer"), + (f64::INFINITY, "integer"), + ] { + let rows = [("api", "checkout", 1.0), ("api", "checkout", weight)]; + let error = try_run(weighted(non_negative()), None, vec![batch(&rows)]) + .err() + .unwrap_or_else(|| panic!("{weight} accepted")); + assert!(error.contains(expected), "{weight}: {error}"); + } +} + +// A group's total (a bare point count) needs a group-mass query that +// asap_sketchlib's Hydra does not offer yet, so it is rejected at binding +// rather than computed from sketchlib internals. +#[test] +fn group_total_is_rejected_until_sketchlib_offers_it() { + let state = compile_node(&summary_agg(unit_count(), None), &[input_schema()]) + .unwrap() + .schema(); + let total = SketchStatistic::PointCount { + key: ColumnRef::SampleValue, + value: None, + }; + let error = compile_node(&estimate(3, total, DataType::Int64), &[state]) + .err() + .expect("group total"); + assert!( + error + .to_string() + .contains("HydraCms does not answer a group total"), + "{error}" + ); +} + +// A filtered HydraCms build updates the grid only from rows where the filter +// is true; every group keeps its row, and `batch`, with no matching row, +// reads no checkout. +#[test] +fn filtered_hydra_cms_counts_matching_rows_and_keeps_groups() { + // weight < 5 drops ("batch", "checkout", 7), ("db", "vacuum", 6). + let filter = Predicate(ScalarExpr::Compare { + semantics: ExprSemantics::Sql, + op: CompareOpKind::Lt, + left: Box::new(ScalarExpr::Column(2)), + right: Box::new(ScalarExpr::Literal(ScalarValue::Float64(5.0))), + }); + let points = run(unit_count(), Some(filter), vec![batch(&ROWS)]); + assert_eq!(points.len(), 3); + assert_eq!(points["batch"], 0.0); + assert!(points["api"] >= 2.0 && points["db"] >= 1.0); +} + +// Groups of one build share one grid, so merging two builds over many groups +// merges the grids once and accounts for each grid once: 400 groups fit the +// default memory limit, which 400 grid copies (about 160 KiB each) exceed. +#[test] +fn many_groups_merge_one_grid_pair() { + let jobs: Vec = (0..400).map(|i| format!("job{i}")).collect(); + let rows: Vec<(&str, &str, f64)> = jobs + .iter() + .map(|job| (job.as_str(), "checkout", 1.0)) + .collect(); + // Every group is in both builds, so every group's state merges. + let merged = try_run(unit_count(), None, vec![batch(&rows), batch(&rows)]) + .expect("merge within the memory limit"); + assert_eq!(merged.len(), 400); + assert!(merged.values().all(|point| *point >= 2.0)); +} diff --git a/crates/asap-physical-operators/tests/physical_semantics.rs b/crates/asap-physical-operators/tests/physical_semantics.rs index 4173730fa..5b8e00c17 100644 --- a/crates/asap-physical-operators/tests/physical_semantics.rs +++ b/crates/asap-physical-operators/tests/physical_semantics.rs @@ -864,3 +864,294 @@ fn empty_exact_summary_extrema_agree_with_ordinary_aggregation() { assert!(matches!(rows[0][0], Value::Null)); } } + +/// Rows of `g`, `x` and a nullable Boolean `keep`; `b` has no kept row, +/// and a NULL `keep` drops its row as SQL FILTER does. +fn filtered_input() -> (SchemaRef, Vec>) { + let input = schema(&[ + ("g", DataType::Utf8, false), + ("x", DataType::Float64, false), + ("keep", DataType::Bool, true), + ]); + let rows = [ + ("a", 1.0, Some(true)), + ("a", 2.0, None), + ("b", 3.0, Some(false)), + ("c", 4.0, Some(true)), + ] + .into_iter() + .map(|(g, x, keep)| { + vec![ + Value::Utf8(g.into()), + Value::Float64(x), + keep.map_or(Value::Null, Value::Bool), + ] + }) + .collect(); + (input, rows) +} + +fn printed(rows: &[Vec]) -> Vec { + let mut rows: Vec<_> = rows + .iter() + .map(|row| { + row.iter() + .map(|v| match v { + Value::Utf8(s) => s.to_string(), + Value::Int64(n) => n.to_string(), + Value::Float64(x) => format!("{x:?}"), + Value::Null => "NULL".into(), + other => panic!("unexpected value {other:?}"), + }) + .collect::>() + .join(" ") + }) + .collect(); + rows.sort(); + rows +} + +fn keep() -> Option> { + Some(Predicate(ScalarExpr::Column(2))) +} + +fn planner_node( + payload: PhysicalASAPOperatorPayload, + fields: &[(&str, FieldDataType, bool)], +) -> PhysicalASAPDAGNode { + PhysicalASAPDAGNode { + id: 1, + payload, + output_state: planner_types::post_asap::ExecutionDataState::QUERY_ROWS, + materialization: Default::default(), + output_schema: planner_types::pre_asap::Schema::new( + fields + .iter() + .map(|(name, dtype, nullable)| Field { + table: None, + name: (*name).into(), + dtype: dtype.clone(), + nullable: *nullable, + }) + .collect(), + ), + guarantee: None, + } +} + +fn round_trip(operator: Operator) -> Operator { + serde_json::from_slice(&serde_json::to_vec(&operator).unwrap()).unwrap() +} + +/// Per-measure filters keep every group: a filtered COUNT reads 0 and a +/// filtered SUM reads NULL for a group with no kept row, and an unfiltered +/// measure beside them still sees every row. The filters survive +/// serialization. +#[test] +fn measure_filters_keep_groups_without_matching_rows() { + use asap_physical_operators::physical_planner::compile_node; + use planner_types::pre_asap::{AggIntent, GroupKeys, Reduction as Plan}; + use planner_types::types::AccuracyTarget; + let (input, rows) = filtered_input(); + let plain = |t: DataType| FieldDataType::Plain(t); + let node = planner_node( + PhysicalASAPOperatorPayload::NonASAP(NonASAPOp::Aggregate { + reduction: Plan::Reduce(GroupKeys::by(vec![0])), + measures: vec![ + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }, + AggIntent::Sum { col: Some(1) }, + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }, + ], + output_names: vec!["c".into(), "s".into(), "all".into()], + filters: vec![keep(), keep(), None], + having: None, + child: 0, + }), + &[ + ("g", plain(DataType::Utf8), false), + ("c", plain(DataType::Int64), false), + ("s", plain(DataType::Float64), true), + ("all", plain(DataType::Int64), false), + ], + ); + let operator = round_trip(compile_node(&node, std::slice::from_ref(&input)).unwrap()); + assert_eq!( + printed(&unary(input, vec![rows], operator)), + ["a 1 1.0 2", "b 0 NULL 1", "c 1 4.0 1"] + ); +} + +/// A filtered SUM is NULL for a group without a kept row, so a plan that +/// declares it non-null does not bind. +#[test] +fn filtered_sum_declared_non_null_is_rejected() { + use asap_physical_operators::physical_planner::compile_node; + use planner_types::pre_asap::{AggIntent, GroupKeys, Reduction as Plan}; + let (input, _) = filtered_input(); + let node = planner_node( + PhysicalASAPOperatorPayload::NonASAP(NonASAPOp::Aggregate { + reduction: Plan::Reduce(GroupKeys::by(vec![0])), + measures: vec![AggIntent::Sum { col: Some(1) }], + output_names: vec!["s".into()], + filters: vec![keep()], + having: None, + child: 0, + }), + &[ + ("g", FieldDataType::Plain(DataType::Utf8), false), + ("s", FieldDataType::Plain(DataType::Float64), false), + ], + ); + assert!(compile_node(&node, &[input]).is_err()); +} + +/// A filtered `SummaryAgg` updates its state only from kept rows but keeps +/// a state for every group: an exact count reads 0 and a KLL quantile, +/// declared nullable, reads NULL for the group with no kept row. The +/// filter survives serialization. +#[test] +fn filtered_summary_build_keeps_empty_groups() { + use asap_physical_operators::physical_planner::compile_node; + use planner_types::ir::ASAPOp; + use planner_types::post_asap::{ + ExactKind, ExactParams, SketchAlgorithm, SketchKind, SketchParams, SketchStatistic, + SummaryUpdate, + }; + use planner_types::pre_asap::{ColumnRef, GroupKeys, Reduction as Plan}; + let (input, rows) = filtered_input(); + let kll = FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + ), + Default::default(), + ); + let count = FieldDataType::ExactAggregate(ExactKind::Count, ExactParams::Count); + let cases = [ + ( + kll, + ASAPOp::SummaryEstimate { + summary_input: 0, + query: SketchStatistic::Quantile { q: 0.5 }, + }, + (DataType::Float64, true), + ["a 1.0", "b NULL", "c 4.0"], + ), + ( + count, + ASAPOp::FinalizeExactAccumulator { child: 0 }, + (DataType::Int64, false), + ["a 1", "b 0", "c 1"], + ), + ]; + for (family, read, (result, nullable), expected) in cases { + let build = planner_node( + PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { + child: 0, + family: family.clone(), + input: SummaryUpdate::column(ColumnRef::Named("x".into())), + reduction: Plan::Reduce(GroupKeys::by(vec![0])), + grouping: Default::default(), + filter: keep(), + }), + &[ + ("g", FieldDataType::Plain(DataType::Utf8), false), + ("state", family, false), + ], + ); + let build = round_trip(compile_node(&build, std::slice::from_ref(&input)).unwrap()); + let read = planner_node( + PhysicalASAPOperatorPayload::ASAP(read), + &[ + ("g", FieldDataType::Plain(DataType::Utf8), false), + ("value", FieldDataType::Plain(result), nullable), + ], + ); + let read = compile_node(&read, &[build.schema()]).unwrap(); + let mut dag = PhysicalDAG::default(); + dag.add( + 0, + vec![], + Operator::source( + input.clone(), + vec![Batch::try_new(input.clone(), rows.clone()).unwrap()], + ) + .unwrap(), + ) + .unwrap(); + dag.add(1, vec![0], build).unwrap(); + dag.add(2, vec![1], read).unwrap(); + assert_eq!(printed(&collect(&dag, 2)), expected); + } +} + +/// A filtered `SummaryAgg` timed at ingestion does not bind: only a +/// query-time build applies its filter. The same build unfiltered binds. +#[test] +fn filtered_summary_build_is_rejected_at_ingestion_time() { + use asap_physical_operators::physical_planner::compile_node; + use planner_types::ir::ASAPOp; + use planner_types::post_asap::{ExactKind, ExactParams, ExecutionDataState, SummaryUpdate}; + use planner_types::pre_asap::{ColumnRef, GroupKeys, Reduction as Plan}; + let (input, _) = filtered_input(); + let family = FieldDataType::ExactAggregate(ExactKind::Count, ExactParams::Count); + let build = |filter| { + let mut node = planner_node( + PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { + child: 0, + family: family.clone(), + input: SummaryUpdate::column(ColumnRef::Named("x".into())), + reduction: Plan::Reduce(GroupKeys::by(vec![0])), + grouping: Default::default(), + filter, + }), + &[ + ("g", FieldDataType::Plain(DataType::Utf8), false), + ("state", family.clone(), false), + ], + ); + node.output_state = ExecutionDataState::INGESTION_SUMMARY; + compile_node(&node, std::slice::from_ref(&input)) + }; + assert!(build(None).is_ok()); + let error = build(keep()).err().expect("filtered ingestion-time build"); + assert!(error.to_string().contains("only at query time"), "{error}"); +} + +/// PromQL never emits per-measure filters, and over PromQL series identity a +/// filtered measure's empty group would be a NULL its non-nullable output +/// cannot hold, so the filter is rejected there. +#[test] +fn measure_filters_are_rejected_over_promql_series_identity() { + let mut input = (*schema(&[ + ( + planner_types::pre_asap::schema::PROMQL_SERIES_IDENTITY, + DataType::Utf8, + false, + ), + ("x", DataType::Float64, false), + ("keep", DataType::Bool, true), + ])) + .clone(); + input.closed = true; + let input = Arc::new(input); + assert!(input.has_promql_series_identity()); + let aggregate = Operator::aggregate( + input.clone(), + vec![0], + vec![("m".into(), Reduction::Min(1))], + ) + .unwrap(); + let error = aggregate + .with_measure_filters(vec![Some(Expression::Column(2))]) + .err() + .expect("filtered measure over PromQL series identity"); + assert!(error.to_string().contains("PromQL"), "{error}"); +} diff --git a/crates/asap-physical-operators/tests/precompute_population.rs b/crates/asap-physical-operators/tests/precompute_population.rs index 2c56ae4d1..154ffe5f6 100644 --- a/crates/asap-physical-operators/tests/precompute_population.rs +++ b/crates/asap-physical-operators/tests/precompute_population.rs @@ -163,6 +163,19 @@ fn finalized_shared_panes_rebuild_one_global_summary_after_recovery() { precompute::compile(&invalid_grouping, &[0], &[3]).is_err(), "numeric values cannot be reinterpreted as population labels" ); + // A filtered summary update runs only at query time. + let mut filtered = dag.clone(); + let PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { filter, .. }) = + &mut filtered.nodes[3].payload + else { + unreachable!() + }; + *filter = Some(planner_types::ir::Predicate( + planner_types::ir::ScalarExpr::Literal(planner_types::pre_asap::ScalarValue::Boolean( + true, + )), + )); + assert!(precompute::compile(&filtered, &[0], &[3]).is_err()); let program = precompute::compile(&dag, &[0], &[3]).unwrap(); let program = serde_json::from_slice::(&serde_json::to_vec(&program).unwrap()) diff --git a/crates/integration-tests/tests/filtered_aggregates.rs b/crates/integration-tests/tests/filtered_aggregates.rs new file mode 100644 index 000000000..07646b214 --- /dev/null +++ b/crates/integration-tests/tests/filtered_aggregates.rs @@ -0,0 +1,333 @@ +//! Filtered aggregates (`FILTER (WHERE …)`, a per-measure filter, or a +//! filtered `SummaryAgg`) execute with SQL semantics: a row that fails the +//! filter contributes nothing, but its group is kept. +mod physical_common; + +use std::collections::BTreeMap; +use std::rc::Rc; +use std::sync::Arc; + +use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_integration_tests::planning::{logical_candidates, selected_dag}; +use asap_physical_operators::{ + physical_planner::{compile, InputContract, Source}, + runtime::{Limits, RunContext, Scope}, + sources::{DataSources, MemorySource}, + values::{Batch, Value}, +}; +use asap_types::ir::physical_export::PhysicalASAPOperatorPayload; +use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, Predicate}; +use asap_types::post_asap::{ExactKind, FieldDataType, SketchAlgorithm}; +use asap_types::pre_asap::schema::{DataType, Field, Schema}; +use asap_types::types::AccuracyTarget; +use futures::StreamExt; + +fn catalog() -> SqlCatalog { + SqlCatalog::new().with_table( + "events", + Schema::new(vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("g", DataType::Utf8, false), + Field::plain("x", DataType::Float64, false), + Field::plain("y", DataType::Float64, true), + ]), + ) +} + +/// Group `b` has rows but none with `x > 0`, and no non-NULL `y`. +fn rows() -> Vec> { + [ + ("a", 1.0, Some(1.0)), + ("a", -1.0, None), + ("a", 2.0, None), + ("b", -3.0, None), + ("b", 0.0, None), + ("c", 5.0, Some(7.0)), + ] + .into_iter() + .map(|(g, x, y)| { + vec![ + Value::Timestamp(0), + Value::Utf8(g.into()), + Value::Float64(x), + y.map_or(Value::Null, Value::Float64), + ] + }) + .collect() +} + +/// Run the logical DAG `root` over [`rows`]; rows come back shown and sorted. +fn execute(root: &Rc) -> Vec { + let dag = physical_common::compile_physical_asap_dag(root).unwrap(); + let scan = dag + .nodes + .iter() + .find(|node| { + matches!( + &node.payload, + PhysicalASAPOperatorPayload::NonASAP(NonASAPOp::Scan { .. }) + ) + }) + .expect("one scan"); + let PhysicalASAPOperatorPayload::NonASAP(NonASAPOp::Scan { source, .. }) = &scan.payload else { + unreachable!() + }; + let schema = Arc::new(scan.output_schema.clone()); + let plan = compile( + &dag, + BTreeMap::from([(scan.id as u64, InputContract::bounded(schema.clone()))]), + &[dag.roots[0] as u64], + ) + .unwrap(); + let expression = OperatorNode::reachable(root) + .into_iter() + .find(|node| matches!(node.non_asap(), Some(NonASAPOp::Scan { source: s, .. }) if s == source)) + .unwrap(); + let mut sources = DataSources::default(); + sources + .register( + source.clone(), + Arc::new( + MemorySource::new( + schema.clone(), + vec![Batch::try_new(schema.clone(), rows()).unwrap()], + ) + .unwrap(), + ), + ) + .unwrap(); + let bound = plan + .instantiate(BTreeMap::from([( + scan.id as u64, + Box::new(sources.bind(&expression).unwrap()) as Source<'_>, + )])) + .unwrap(); + let context = RunContext::new( + Scope::Query { + evaluation_time_ms: 0, + revision: 1, + }, + Limits::default(), + ) + .unwrap(); + let show = |value: &Value| match value { + Value::Null => "NULL".to_string(), + Value::Int64(value) => value.to_string(), + Value::Float64(value) => format!("{value:?}"), + Value::Utf8(value) => value.to_string(), + other => format!("{other:?}"), + }; + let mut out: Vec = futures::executor::block_on(async { + let mut stream = bound.execute(plan.roots(), context).unwrap().remove(0); + let mut rows = Vec::new(); + while let Some(batch) = stream.next().await { + rows.extend( + batch + .unwrap() + .rows() + .iter() + .map(|row| row.iter().map(show).collect::>().join(" ")), + ); + } + rows + }); + out.sort(); + out +} + +async fn lower(sql: &str, accuracy: AccuracyTarget) -> Rc { + lower_sql(sql, &catalog(), accuracy).await.unwrap() +} + +/// The selected plan for an exact `sql`, through all three stages. +async fn exact_plan(sql: &str) -> Rc { + selected_dag( + lower(sql, AccuracyTarget::Exact).await, + AccuracyTarget::Exact, + ) +} + +/// `COUNT(*) FILTER (WHERE x > 0) GROUP BY g` counts the matching rows per +/// group, and a group without one reports 0. +#[tokio::test] +async fn exact_filtered_count_keeps_groups_without_matches() { + let root = + exact_plan("SELECT g, COUNT(*) FILTER (WHERE x > 0) AS c FROM events GROUP BY g").await; + assert_eq!(execute(&root), ["a 2", "b 0", "c 1"]); +} + +/// `SUM(x) FILTER (WHERE x > 0) GROUP BY g` sums the matching rows; a group +/// without one reports NULL, as SQL's SUM over no rows does. +#[tokio::test] +async fn exact_filtered_sum_is_null_for_groups_without_matches() { + let root = + exact_plan("SELECT g, SUM(x) FILTER (WHERE x > 0) AS s FROM events GROUP BY g").await; + assert_eq!(execute(&root), ["a 3.0", "b NULL", "c 5.0"]); +} + +/// `COUNT(y)` over a nullable `y` is lowered as a count filtered by +/// `y IS NOT NULL`, so it executes the same way. +#[tokio::test] +async fn exact_count_of_nullable_column_skips_nulls() { + let root = exact_plan("SELECT g, COUNT(y) AS c FROM events GROUP BY g").await; + assert_eq!(execute(&root), ["a 1", "b 0", "c 1"]); +} + +/// `root` with each unfiltered `SummaryAgg`'s filter set to `filter`. +fn with_summary_filter(root: &Rc, filter: &Predicate) -> Rc { + if let Operator::ASAP(ASAPOp::SummaryAgg { + child, + family, + input, + reduction, + grouping, + filter: None, + }) = &root.operator + { + return Rc::new( + OperatorNode::new(Operator::ASAP(ASAPOp::SummaryAgg { + child: child.clone(), + family: family.clone(), + input: input.clone(), + reduction: reduction.clone(), + grouping: grouping.clone(), + filter: Some(filter.clone()), + })) + .unwrap(), + ); + } + Rc::new( + root.with_new_children(|c| with_summary_filter(c, filter)) + .unwrap(), + ) +} + +/// The predicate of the filtered aggregate in `sql`, and its input schema. +async fn measure_filter(sql: &str) -> (Predicate, Schema) { + OperatorNode::reachable(&lower(sql, AccuracyTarget::Exact).await) + .into_iter() + .find_map(|n| match n.non_asap() { + Some(NonASAPOp::Aggregate { filters, child, .. }) => { + Some((filters[0].clone()?, child.schema.clone())) + } + _ => None, + }) + .expect("filtered aggregate") +} + +/// The logical candidates for `sql` whose single `SummaryAgg` has a family +/// matching `family`, with that `SummaryAgg`'s filter set by hand. +async fn hand_filtered( + sql: &str, + family: impl Fn(&FieldDataType) -> bool, + filter: &(Predicate, Schema), +) -> Vec> { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + let mut result = vec![]; + for root in logical_candidates(lower(sql, target.clone()).await, target) { + let builds: Vec<_> = OperatorNode::reachable(&root) + .into_iter() + .filter(|n| matches!(n.asap(), Some(ASAPOp::SummaryAgg { .. }))) + .collect(); + let [build] = builds.as_slice() else { + continue; + }; + let Some(ASAPOp::SummaryAgg { + family: actual, + child, + .. + }) = build.asap() + else { + unreachable!() + }; + if family(actual) { + // The filter's column indices refer to the same input relation. + assert_eq!(child.schema, filter.1); + result.push(with_summary_filter(&root, &filter.0)); + } + } + result +} + +/// A hand-filtered exact `Count` accumulator counts only `x > 0` per `g` +/// but keeps every group: `b`, with no matching row, reads 0. +#[tokio::test] +async fn hand_filtered_exact_count_compiles_and_executes() { + let filter = + measure_filter("SELECT g, COUNT(*) FILTER (WHERE x > 0) AS c FROM events GROUP BY g").await; + let roots = hand_filtered( + "SELECT g, COUNT(*) AS c FROM events GROUP BY g", + |f| matches!(f, FieldDataType::ExactAggregate(ExactKind::Count, _)), + &filter, + ) + .await; + assert!(!roots.is_empty()); + for root in roots { + assert_eq!(execute(&root), ["a 2", "b 0", "c 1"]); + } +} + +/// A hand-filtered KLL sketch holds only the `x >= 0` values of each `g` +/// and keeps every group. (The unfiltered plan declares a non-null result, +/// so no group is left empty here; `physical_semantics` covers NULL.) +#[tokio::test] +async fn hand_filtered_kll_compiles_and_executes() { + let filter = measure_filter( + "SELECT g, approx_percentile_cont(x, 0.5) FILTER (WHERE x >= 0) AS q FROM events GROUP BY g", + ) + .await; + let roots = hand_filtered( + "SELECT g, approx_percentile_cont(x, 0.5) AS q FROM events GROUP BY g", + |f| matches!(f, FieldDataType::Sketch(kind, _) if kind.algorithm() == &SketchAlgorithm::Kll), + &filter, + ) + .await; + assert!(!roots.is_empty()); + for root in roots { + assert_eq!(execute(&root), ["a 1.0", "b 0.0", "c 5.0"]); + } +} + +/// The KLL candidates of `sql`, unchanged. +async fn kll_candidates(sql: &str) -> Vec> { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + let roots: Vec<_> = logical_candidates(lower(sql, target.clone()).await, target) + .into_iter() + .filter(|root| { + OperatorNode::reachable(root).iter().any(|n| { + matches!(n.asap(), Some(ASAPOp::SummaryAgg { family: FieldDataType::Sketch(kind, _), .. }) + if kind.algorithm() == &SketchAlgorithm::Kll) + }) + }) + .collect(); + assert!(!roots.is_empty(), "no KLL candidate for {sql}"); + roots +} + +/// A KLL quantile of a nullable column is NULL for a group whose values are +/// all NULL (`b`), as SQL's quantile over no value is, instead of failing +/// the query. +#[tokio::test] +async fn kll_quantile_of_all_null_group_is_null() { + for root in + kll_candidates("SELECT g, approx_percentile_cont(y, 0.5) AS q FROM events GROUP BY g").await + { + assert_eq!(execute(&root), ["a 1.0", "b NULL", "c 7.0"]); + } +} + +/// An ungrouped KLL quantile over no row is one NULL row. +#[tokio::test] +async fn global_kll_quantile_over_no_row_is_null() { + for root in + kll_candidates("SELECT approx_percentile_cont(x, 0.5) AS q FROM events WHERE x > 100").await + { + assert_eq!(execute(&root), ["NULL"]); + } +} diff --git a/crates/types/src/ir/asap.rs b/crates/types/src/ir/asap.rs index 902616ea8..75f8d1018 100644 --- a/crates/types/src/ir/asap.rs +++ b/crates/types/src/ir/asap.rs @@ -270,16 +270,8 @@ impl ASAPOp { { dtype = DataType::Float64; } - // A filtered quantile is NULL for a group without a matching - // row, as the filtered aggregate it realizes is. let nullable = matches!(query, SketchStatistic::Quantile { .. }) - && matches!( - summary_input.asap(), - Some(SummaryAgg { - filter: Some(_), - .. - }) - ); + && quantile_may_be_empty(summary_input); let mut schema = input.clone(); for field in &mut schema.fields { if !field.is_plain() { @@ -654,3 +646,42 @@ fn source_kind(node: &OperatorNode) -> OperatorResultKind { }, } } + +/// Whether a quantile read from `state` can have no value to rank, and so is +/// NULL as the SQL aggregate it realizes is: the `SummaryAgg` under it (seen +/// through `SummaryMerge`) is filtered, or, over a relation, is global (zero +/// rows give one empty state) or summarizes a nullable column (a group of +/// NULLs gives an empty state). +fn quantile_may_be_empty(state: &OperatorNode) -> bool { + match state.asap() { + Some(ASAPOp::SummaryMerge { children }) => { + children.iter().any(|child| quantile_may_be_empty(child)) + } + Some(ASAPOp::SummaryAgg { + child, + input, + reduction, + filter, + .. + }) => { + if filter.is_some() { + return true; + } + if child.result_kind != OperatorResultKind::Relation { + return false; + } + let global = matches!(reduction, Reduction::Reduce(by) if by.keys().is_empty()); + let nullable = match &input.weight { + crate::post_asap::sketch::SummaryInputExpr::Column(column) => { + crate::pre_asap::resolve_column_ref(column, &child.schema) + .ok() + .and_then(|index| child.schema.fields.get(index)) + .is_some_and(|field| field.nullable) + } + _ => false, + }; + global || nullable + } + _ => false, + } +} diff --git a/crates/types/src/ir/non_asap.rs b/crates/types/src/ir/non_asap.rs index 38b1606c5..bb05197e8 100644 --- a/crates/types/src/ir/non_asap.rs +++ b/crates/types/src/ir/non_asap.rs @@ -529,6 +529,23 @@ impl NonASAPOp { } } } + // Over a relation, a quantile is also NULL for a global + // aggregate of no row and a group whose values are all NULL. + if child.result_kind == OperatorResultKind::Relation { + for (index, measure) in measures.iter().enumerate() { + if matches!(measure, AggIntent::Quantile { .. }) + && (offset == 0 + || measure + .input_cols() + .iter() + .any(|i| child.schema.fields[*i].nullable)) + { + if let Some(field) = output.fields.get_mut(offset + index) { + field.nullable = true; + } + } + } + } if child.result_kind == OperatorResultKind::Relation { for (index, measure) in measures.iter().enumerate() { if matches!(