diff --git a/crates/executor/src/operators/aggregate/mod.rs b/crates/executor/src/operators/aggregate/mod.rs index e2b67442..9d64b86f 100644 --- a/crates/executor/src/operators/aggregate/mod.rs +++ b/crates/executor/src/operators/aggregate/mod.rs @@ -83,11 +83,50 @@ 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")); + } + 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(_) + ) && !self.inputs[0].has_promql_series_identity() + { + output.fields[groups.len() + index].nullable = true; + } + } + *slot = filters; + self.output = Arc::new(output); + Ok(self) + } pub fn window( input: SchemaRef, intent: planner_types::ir::operator::AggIntent, @@ -189,7 +228,7 @@ pub(super) fn execute<'a>( Kind::SQLWindowSum { column } => { let mut work = Cooperative::new(&context); let total = reduce_one( - &rows, + &rows.iter().collect::>(), &Reduction::Sum(*column), &operator.inputs[0], &mut work, @@ -224,8 +263,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!(), }; @@ -239,6 +290,7 @@ async fn reduce( rows: Vec>, groups: &[usize], measures: &[Reduction], + filters: &[Option], input: &SchemaRef, context: &RunContext, ) -> Result>, Error> { @@ -265,8 +317,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, context).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, context).await?); } workspace.grow(row_bytes(&result))?; output.push(result); @@ -300,7 +365,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/executor/src/operators/mod.rs b/crates/executor/src/operators/mod.rs index cec05d79..de802442 100644 --- a/crates/executor/src/operators/mod.rs +++ b/crates/executor/src/operators/mod.rs @@ -126,6 +126,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)>, @@ -141,12 +144,18 @@ enum Kind { value: Option, 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 shared state for all groups (HydraCms); `weight: None` is a unit count. SharedSummaryBuild { @@ -154,6 +163,9 @@ enum Kind { item: usize, 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, diff --git a/crates/executor/src/operators/summary/mod.rs b/crates/executor/src/operators/summary/mod.rs index bb57b343..3918b4fd 100644 --- a/crates/executor/src/operators/summary/mod.rs +++ b/crates/executor/src/operators/summary/mod.rs @@ -55,6 +55,7 @@ impl Operator { value, items, groups, + filter: None, }, inputs: vec![input], output: schema(fields), @@ -98,6 +99,7 @@ impl Operator { item, weight, groups, + filter: None, }, inputs: vec![input], output: schema(fields), @@ -216,11 +218,27 @@ impl Operator { value, time, 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 summary_merge( input: SchemaRef, state: usize, @@ -314,10 +332,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()), @@ -326,10 +345,12 @@ 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()), @@ -338,10 +359,12 @@ pub(super) fn execute<'a>( item, weight, groups, + filter, } => Ok(futures::stream::once(async move { Batch::try_new( output, - build_shared_summary(input, family, *item, *weight, groups, &context).await?, + build_shared_summary(input, family, *item, *weight, groups, filter.as_deref(), &context) + .await?, ) }) .boxed_local()), @@ -388,6 +411,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) @@ -462,12 +491,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: Option, time: Option, groups: &[usize], + filter: Option<&Expression>, emit_empty_global: bool, context: &RunContext, ) -> Result>, Error> { @@ -518,6 +549,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. @@ -561,6 +595,13 @@ 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( rows: Vec>, state_column: usize, @@ -630,6 +671,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}; @@ -666,6 +708,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")); @@ -701,6 +746,7 @@ async fn build_shared_summary( item: usize, weight: Option, groups: &[usize], + filter: Option<&Expression>, context: &RunContext, ) -> Result>, Error> { use crate::summary_kernels::HydraCmsGroup; @@ -729,6 +775,9 @@ async fn build_shared_summary( memory.resize(retained)?; seen.insert(key.clone(), (labels, name)); } + if !selected(filter, row)? { + continue; + } let count = match weight.map(|column| &row[column]) { None => 1, // SQL aggregates ignore NULL weights while retaining the group. diff --git a/crates/executor/src/operators/unchecked.rs b/crates/executor/src/operators/unchecked.rs index 6061a377..fae4cc4f 100644 --- a/crates/executor/src/operators/unchecked.rs +++ b/crates/executor/src/operators/unchecked.rs @@ -125,12 +125,17 @@ impl TryFrom for Operator { .clone(); Operator::sql_window_sum(input(0)?, column, name)? } - 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, @@ -155,22 +160,34 @@ impl TryFrom for Operator { value, time, groups, - } => match value { - Some(value) => Operator::summary_build(input(0)?, family, value, time, groups)?, - None => Operator::unit_count_build(input(0)?, family, groups)?, - }, + filter, + } => with_filter( + match value { + Some(value) => Operator::summary_build(input(0)?, family, value, time, groups)?, + None => Operator::unit_count_build(input(0)?, family, 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, - } => Operator::shared_summary_build(input(0)?, 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())? } @@ -191,3 +208,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/executor/src/physical_planner/mod.rs b/crates/executor/src/physical_planner/mod.rs index 2491942c..c2389af7 100644 --- a/crates/executor/src/physical_planner/mod.rs +++ b/crates/executor/src/physical_planner/mod.rs @@ -860,9 +860,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")); } @@ -907,7 +904,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")), }, @@ -918,106 +921,11 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result { - if filter.is_some() { - return Err(invalid( - "filtered summary update has no native implementation", - )); + let build = summary_build(input, family, update, reduction, grouping)?; + match filter { + Some(filter) => build.with_row_filter(expression(&filter.0, input)?), + None => Ok(build), } - if grouping != &planner_types::ir::schema::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") - }; - 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::ir::schema::SketchAlgorithm::CmsWithHeap) - && !matches!( - update.weight_domain, - planner_types::ir::schema::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 PlannerReduction::Reduce(keys) = reduction else { - return Err(invalid( - "summary construction requires explicit grouping columns", - )); - }; - let column = match &update.weight { - SummaryInputExpr::Column(column) => column, - SummaryInputExpr::Constant(weight) if *weight == 1.0 => { - return Operator::unit_count_build( - input.clone(), - family.clone(), - groups(input, keys)?, - ) - } - _ => { - return Err(invalid( - "summary update expression must be projected to a column", - )) - } - }; - Operator::summary_build( - input.clone(), - family.clone(), - named_column(input, column)?, - input.time_index, - groups(input, keys)?, - ) } Payload::SummaryMerge => { let state = summary_column(input)?; @@ -1049,6 +957,104 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result Result { + if grouping != &planner_types::ir::schema::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") + }; + 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::ir::schema::SketchAlgorithm::CmsWithHeap) + && !matches!( + update.weight_domain, + planner_types::ir::schema::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 PlannerReduction::Reduce(keys) = reduction else { + return Err(invalid( + "summary construction requires explicit grouping columns", + )); + }; + let column = match &update.weight { + SummaryInputExpr::Column(column) => column, + SummaryInputExpr::Constant(weight) if *weight == 1.0 => { + return Operator::unit_count_build(input.clone(), family.clone(), groups(input, keys)?) + } + _ => { + return Err(invalid( + "summary update expression must be projected to a column", + )) + } + }; + 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/executor/src/summary_kernels/datasketches_kll.rs b/crates/executor/src/summary_kernels/datasketches_kll.rs index d6460706..b070ca5e 100644 --- a/crates/executor/src/summary_kernels/datasketches_kll.rs +++ b/crates/executor/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/executor/src/summary_kernels/dd_sketch.rs b/crates/executor/src/summary_kernels/dd_sketch.rs index 72f786ef..735effce 100644 --- a/crates/executor/src/summary_kernels/dd_sketch.rs +++ b/crates/executor/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/executor/src/summary_kernels/traits.rs b/crates/executor/src/summary_kernels/traits.rs index 8fd0b52d..9a9f2e1a 100644 --- a/crates/executor/src/summary_kernels/traits.rs +++ b/crates/executor/src/summary_kernels/traits.rs @@ -26,6 +26,13 @@ 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. fn approx_memory_bytes(&self) -> usize { 4096 diff --git a/crates/executor/tests/physical_semantics.rs b/crates/executor/tests/physical_semantics.rs index 3f8538de..9f903b54 100644 --- a/crates/executor/tests/physical_semantics.rs +++ b/crates/executor/tests/physical_semantics.rs @@ -1001,3 +1001,126 @@ fn sql_ln_executes_numeric_and_null_arguments() { )); } } + +/// 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 +} + +/// Per-measure filters keep every group: a filtered COUNT reads 0 and a +/// filtered SUM reads NULL (declared nullable) 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() { + let (input, rows) = filtered_input(); + let operator = Operator::aggregate( + input.clone(), + vec![0], + vec![ + ("c".into(), Reduction::Count), + ("s".into(), Reduction::Sum(1)), + ("all".into(), Reduction::Count), + ], + ) + .unwrap() + .with_measure_filters(vec![ + Some(Expression::Column(2)), + Some(Expression::Column(2)), + None, + ]) + .unwrap(); + assert!(operator.schema().fields[2].nullable); + assert!(!operator.schema().fields[1].nullable); + let operator: Operator = + serde_json::from_slice(&serde_json::to_vec(&operator).unwrap()).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 KLL build keeps a state for every group; the group with no +/// kept row reads NULL when its result is declared nullable. The filter +/// survives serialization. +#[test] +fn filtered_summary_build_keeps_empty_groups() { + use planner_types::ir::schema::{SketchAlgorithm, SketchKind, SketchParams}; + let (input, rows) = filtered_input(); + let family = FieldDataType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + Default::default(), + ); + let build = Operator::summary_build(input.clone(), family, 1, None, vec![0]) + .unwrap() + .with_row_filter(Expression::Column(2)) + .unwrap(); + let build: Operator = serde_json::from_slice(&serde_json::to_vec(&build).unwrap()).unwrap(); + let evaluation = Operator::evaluation( + build.schema(), + 1, + asap_executor::operators::SummaryEvaluation::Sketch( + planner_types::ir::schema::SketchStatistic::Quantile { q: 0.5 }, + ), + ) + .unwrap(); + // Declare the result nullable, as the Planner does for a filtered quantile. + let mut wire = serde_json::to_value(&evaluation).unwrap(); + wire["output"]["fields"][1]["nullable"] = true.into(); + let evaluation: Operator = serde_json::from_value(wire).unwrap(); + let mut dag = PhysicalDAG::default(); + dag.add( + 0, + vec![], + Operator::source( + input.clone(), + vec![Batch::try_new(input.clone(), rows).unwrap()], + ) + .unwrap(), + ) + .unwrap(); + dag.add(1, vec![0], build).unwrap(); + dag.add(2, vec![1], evaluation).unwrap(); + assert_eq!(printed(&collect(&dag, 2)), ["a 1.0", "b NULL", "c 4.0"]); +} diff --git a/crates/integration-tests/tests/filtered_aggregates.rs b/crates/integration-tests/tests/filtered_aggregates.rs new file mode 100644 index 00000000..817e9263 --- /dev/null +++ b/crates/integration-tests/tests/filtered_aggregates.rs @@ -0,0 +1,330 @@ +//! 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 asap_executor::values::Value; +use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_logical_optimizer::pass1::logical_candidates::{ + compose_logical_candidate, enumerate_choices, enumerate_local_logical_candidates, +}; +use asap_plan_selection::{plan_stages, PlanningModels}; +use asap_types::ir::schema::{DataType, Field, Schema}; +use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, QueryRoot}; +use asap_types::types::AccuracyTarget; +use asap_types::workload::{ + DataArrival, DataWorkload, Evidence, EvidenceSource, Predictability, QueryRecurrence, Rate, + RootDemand, +}; + +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() +} + +/// The executed rows, printed and sorted (`Value` has no `PartialEq`). +fn sorted(root: &Rc) -> Vec { + let mut rows: Vec<_> = physical_common::execute_raw_rows(root, rows()) + .iter() + .map(|row| print(row)) + .collect(); + rows.sort(); + rows +} + +fn print(row: &[Value]) -> String { + 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(" ") +} + +fn printed(rows: [Vec; 3]) -> Vec { + rows.iter().map(|row| print(row)).collect() +} + +fn declared(value: T) -> Evidence { + Evidence { + value: Some(value), + source: EvidenceSource::Declared, + ..Default::default() + } +} + +/// The selected plan's root for an exact `sql`, through all three stages. +async fn selected(sql: &str) -> Rc { + let root = lower_sql(sql, &catalog(), AccuracyTarget::Exact) + .await + .unwrap(); + let demand = [RootDemand { + accuracy: Some(AccuracyTarget::Exact), + recurrence: QueryRecurrence::OneTime { + invocations: 1, + execute_at: None, + }, + predictability: Predictability::default(), + latency_ms: None, + }]; + let data = DataWorkload { + arrival: DataArrival::ContinuouslyIngesting, + ingestion_rate: declared(Rate(1_000.0)), + input_cardinality: declared(1_000_000), + ..Default::default() + }; + let run = plan_stages( + vec![(0, QueryRoot::Operator(root))], + &demand, + &data, + PlanningModels::builtin(), + 4096, + ) + .unwrap(); + let QueryRoot::Operator(root) = &run.plan.logical[0].1 else { + panic!("operator root") + }; + root.clone() +} + +fn s(v: &str) -> Value { + Value::Utf8(v.into()) +} + +/// `COUNT(*) FILTER (WHERE x > 0) GROUP BY g` counts the matching rows per +/// group, and a group without one reports 0, through `plan_stages` and the +/// executor. +#[tokio::test] +async fn exact_filtered_count_keeps_groups_without_matches() { + let root = + selected("SELECT g, COUNT(*) FILTER (WHERE x > 0) AS c FROM events GROUP BY g").await; + assert_eq!( + sorted(&root), + printed([ + vec![s("a"), Value::Int64(2)], + vec![s("b"), Value::Int64(0)], + vec![s("c"), Value::Int64(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 = selected("SELECT g, SUM(x) FILTER (WHERE x > 0) AS s FROM events GROUP BY g").await; + assert_eq!( + sorted(&root), + printed([ + vec![s("a"), Value::Float64(3.0)], + vec![s("b"), Value::Null], + vec![s("c"), Value::Float64(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 = selected("SELECT g, COUNT(y) AS c FROM events GROUP BY g").await; + assert_eq!( + sorted(&root), + printed([ + vec![s("a"), Value::Int64(1)], + vec![s("b"), Value::Int64(0)], + vec![s("c"), Value::Int64(1)], + ]) + ); +} + +/// The `SummaryAgg` nodes in `root`. +fn summary_builds(root: &Rc) -> Vec> { + OperatorNode::reachable(root) + .into_iter() + .filter(|n| matches!(n.asap(), Some(ASAPOp::SummaryAgg { .. }))) + .collect() +} + +/// `root` with each `SummaryAgg`'s filter set to `filter`. +fn with_summary_filter( + root: &Rc, + filter: &asap_types::ir::scalar::Predicate, +) -> Rc { + if let Operator::ASAP(ASAPOp::SummaryAgg { + child, + family, + input, + reduction, + grouping, + filter: None, + }) = &root.operator + { + let state = 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(); + let state = match &root.coverage { + Some(coverage) => state.with_coverage(coverage.clone()).unwrap(), + None => state, + }; + return Rc::new(state); + } + Rc::new( + root.map_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) -> (asap_types::ir::scalar::Predicate, Schema) { + let root = lower_sql(sql, &catalog(), AccuracyTarget::Exact) + .await + .unwrap(); + OperatorNode::reachable(&root) + .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") +} + +/// Pass 1's alternatives for `sql` whose single `SummaryAgg` has `family`, +/// with that `SummaryAgg`'s filter set to `filter` by hand. +async fn hand_filtered( + sql: &str, + family: impl Fn(&asap_types::ir::schema::FieldDataType) -> bool, + filter: &(asap_types::ir::scalar::Predicate, Schema), +) -> Vec> { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + let root = lower_sql(sql, &catalog(), target).await.unwrap(); + let inventory = + enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))], &BTreeMap::new()) + .unwrap(); + let mut result = vec![]; + for choice in enumerate_choices(&inventory, usize::MAX) { + let roots = compose_logical_candidate(&inventory, &choice).unwrap(); + let QueryRoot::Operator(root) = &roots[0].1 else { + panic!("operator root") + }; + let builds = summary_builds(root); + 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() { + use asap_types::ir::schema::{ExactKind, FieldDataType}; + 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_eq!(roots.len(), 1); + physical_common::compile_physical_asap_dag(&roots[0]).unwrap(); + assert_eq!( + sorted(&roots[0]), + printed([ + vec![s("a"), Value::Int64(2)], + vec![s("b"), Value::Int64(0)], + vec![s("c"), Value::Int64(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() { + use asap_types::ir::schema::{FieldDataType, SketchAlgorithm}; + 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 { + physical_common::compile_physical_asap_dag(&root).unwrap(); + assert_eq!( + sorted(&root), + printed([ + vec![s("a"), Value::Float64(1.0)], + vec![s("b"), Value::Float64(0.0)], + vec![s("c"), Value::Float64(5.0)], + ]) + ); + } +} diff --git a/crates/types/src/ir/operator/non_asap.rs b/crates/types/src/ir/operator/non_asap.rs index 3b71c3bf..d748a615 100644 --- a/crates/types/src/ir/operator/non_asap.rs +++ b/crates/types/src/ir/operator/non_asap.rs @@ -548,6 +548,14 @@ impl NonASAPOp { field.nullable = nullable; } } + // A quantile over no matching row is NULL. + if matches!(measure, AggIntent::Quantile { .. }) + && filters.get(index).is_some_and(Option::is_some) + { + if let Some(field) = output.fields.get_mut(offset + index) { + field.nullable = true; + } + } } } output