diff --git a/datafusion/functions-aggregate/src/first_last.rs b/datafusion/functions-aggregate/src/first_last.rs index cecb277cb844a..7c2540aa03a2b 100644 --- a/datafusion/functions-aggregate/src/first_last.rs +++ b/datafusion/functions-aggregate/src/first_last.rs @@ -555,8 +555,15 @@ impl FirstLastGroupsAccumulator { for (idx_in_val, group_idx) in group_indices.iter().enumerate() { let group_idx = *group_idx; - let passed_filter = opt_filter.is_none_or(|x| x.value(idx_in_val)); - let is_set = is_set_arr.is_none_or(|x| x.value(idx_in_val)); + // A row passes the FILTER clause only when the predicate is + // `true`; rows whose predicate evaluates to `null` are excluded. + let passed_filter = + opt_filter.is_none_or(|x| x.is_valid(idx_in_val) && x.value(idx_in_val)); + // `is_set_arr` carries the user FILTER clause (including its + // nulls) when the state was produced by `convert_to_state`, so + // the validity check is required here as well (#22666). + let is_set = + is_set_arr.is_none_or(|x| x.is_valid(idx_in_val) && x.value(idx_in_val)); if !passed_filter || !is_set { continue; @@ -1415,6 +1422,7 @@ mod tests { use arrow::{ array::{BooleanArray, Int64Array, ListArray, PrimitiveArray, StringArray}, + buffer::NullBuffer, compute::SortOptions, datatypes::Schema, }; @@ -1773,6 +1781,118 @@ mod tests { Ok(()) } + /// Rows whose FILTER predicate evaluates to `null` must not pass the + /// filter, even when the underlying value bit at the null slot is `true` + /// (#22666). + #[test] + fn test_group_acc_filter_null_predicate() -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, true), + Field::new("c", DataType::Int64, true), + ])); + + let sort_keys = [PhysicalSortExpr { + expr: col("c", &schema).unwrap(), + options: SortOptions::default(), + }]; + + let mut group_acc = FirstLastGroupsAccumulator::try_new( + PrimitiveValueState::::new(DataType::Int64), + sort_keys.into(), + true, + &[DataType::Int64], + true, + )?; + + let val_with_orderings: Vec = vec![ + Arc::new(Int64Array::from(vec![10, 20, 30])), + Arc::new(Int64Array::from(vec![10, 20, 30])), + ]; + + // Row 0: predicate is null (but its value bit is true, as produced by + // kernels such as `b < 1` when the null slot's underlying value is 0) + // Row 1: predicate is false + // Row 2: predicate is true + let filter = BooleanArray::new( + BooleanBuffer::from(vec![false, true, false, true]), + Some(NullBuffer::from(BooleanBuffer::from(vec![ + true, false, true, true, + ]))), + ) + .slice(1, 3); + assert_eq!(filter.offset(), 1); + + group_acc.update_batch(&val_with_orderings, &[0, 0, 1], Some(&filter), 2)?; + + let binding = group_acc.evaluate(EmitTo::All)?; + let eval_result = binding.as_any().downcast_ref::().unwrap(); + + // Group 0 has no row with a `true` predicate, so it must stay unset. + // Group 1 takes the only row with a `true` predicate. + let expect: PrimitiveArray = Int64Array::from(vec![None, Some(30)]); + assert_eq!(eval_result, &expect); + + Ok(()) + } + + /// `convert_to_state` stores the user FILTER clause (including its nulls) + /// in the `is_set` state column, so `merge_batch` must not treat a null + /// `is_set` entry with a set value bit as "is set" (#22666). + #[test] + fn test_group_acc_merge_null_is_set() -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, true), + Field::new("c", DataType::Int64, true), + ])); + + let sort_keys = [PhysicalSortExpr { + expr: col("c", &schema).unwrap(), + options: SortOptions::default(), + }]; + + let group_acc = FirstLastGroupsAccumulator::try_new( + PrimitiveValueState::::new(DataType::Int64), + sort_keys.clone().into(), + true, + &[DataType::Int64], + true, + )?; + + let val_with_orderings: Vec = vec![ + Arc::new(Int64Array::from(vec![10, 20])), + Arc::new(Int64Array::from(vec![10, 20])), + ]; + + // Same null-with-set-value-bit filter as above, carried into the state + let filter = BooleanArray::new( + BooleanBuffer::from(vec![true, true]), + Some(NullBuffer::from(BooleanBuffer::from(vec![false, true]))), + ); + + let state = group_acc.convert_to_state(&val_with_orderings, Some(&filter))?; + assert_eq!(state.len(), 3); + + let mut merging_acc = FirstLastGroupsAccumulator::try_new( + PrimitiveValueState::::new(DataType::Int64), + sort_keys.into(), + true, + &[DataType::Int64], + true, + )?; + + merging_acc.merge_batch(&state, &[0, 0], 1)?; + + let binding = merging_acc.evaluate(EmitTo::All)?; + let eval_result = binding.as_any().downcast_ref::().unwrap(); + + // Only the second row is valid and passes; the null-predicate row must + // be skipped even though its value bit is true. + let expect: PrimitiveArray = Int64Array::from(vec![Some(20)]); + assert_eq!(eval_result, &expect); + + Ok(()) + } + #[test] fn test_first_list_acc_size() -> Result<()> { fn size_after_batch(values: &[ArrayRef]) -> Result { diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt index cbb9c5d0317dc..bee56040cd160 100644 --- a/datafusion/sqllogictest/test_files/aggregate.slt +++ b/datafusion/sqllogictest/test_files/aggregate.slt @@ -6659,6 +6659,69 @@ GROUP BY g ---- 0 0 +# first_value_with_group_by_and_nullable_filter +# Rows whose FILTER predicate evaluates to NULL must be excluded (#22666) +query II rowsort +SELECT g, first_value(a ORDER BY a) FILTER (WHERE b < 1) AS fv +FROM (VALUES (0, 10, CAST(NULL AS INT)), (0, 20, 2)) AS t(g, a, b) +GROUP BY g +---- +0 NULL + +# last_value_with_group_by_and_nullable_filter +query II rowsort +SELECT g, last_value(a ORDER BY a) FILTER (WHERE b < 1) AS lv +FROM (VALUES (0, 10, CAST(NULL AS INT)), (0, 20, 2)) AS t(g, a, b) +GROUP BY g +---- +0 NULL + +# first_last_value_with_group_by_and_mixed_filter_results +# Only rows whose FILTER predicate is TRUE participate: a = 10 (b = 1) and +# a = 20 (b = 0) in group 0. The NULL-predicate row (a = 5) and the +# FALSE-predicate row (a = 30) are excluded. No row passes the filter in +# group 1, so the aggregates return NULL there. +query III rowsort +SELECT g, + first_value(a ORDER BY a) FILTER (WHERE b < 2) AS fv, + last_value(a ORDER BY a) FILTER (WHERE b < 2) AS lv +FROM (VALUES (0, 5, CAST(NULL AS INT)), (0, 10, 1), (0, 30, 2), (0, 20, 0), + (1, 100, CAST(NULL AS INT)), (1, 50, 3)) AS t(g, a, b) +GROUP BY g +---- +0 10 20 +1 NULL NULL + +# first_last_value_with_group_by_filter_all_true_and_no_filter +# Behavior is unchanged when every row passes the FILTER or there is no FILTER +query IIIII rowsort +SELECT g, + first_value(a ORDER BY a) FILTER (WHERE a > 0) AS fv, + last_value(a ORDER BY a) FILTER (WHERE a > 0) AS lv, + first_value(a ORDER BY a) AS fv_no_filter, + last_value(a ORDER BY a) AS lv_no_filter +FROM (VALUES (0, 5, CAST(NULL AS INT)), (0, 10, 1), (0, 30, 2), (0, 20, 0)) AS t(g, a, b) +GROUP BY g +---- +0 5 30 5 30 + +# first_value_without_group_by_and_nullable_filter +query I rowsort +SELECT first_value(a ORDER BY a) FILTER (WHERE b < 1) AS fv +FROM (VALUES (10, CAST(NULL AS INT)), (20, 2)) AS t(a, b) +---- +NULL + +# first_value_window_function_no_regression +query II +SELECT a, first_value(a) OVER (ORDER BY a) AS fv +FROM (VALUES (10), (20), (5)) AS t(a) +ORDER BY a +---- +5 5 +10 5 +20 5 + # query_with_untyped_null_filter query I SELECT count(*) FILTER (WHERE NULL)