This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23707-f40d99ac8b10e03a41374706e9fa07194a922ca9 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 18b1e359c3c547ec0d649932f85abbe33144ef19 Author: kid <[email protected]> AuthorDate: Fri Jul 24 17:15:20 2026 +0800 fix: grouped first_value/last_value FILTER excludes NULL predicate rows (#23707) ## Which issue does this PR close? - Closes #22666 ## Rationale for this change Under SQL aggregate `FILTER` semantics, a row passes only when the predicate evaluates to `true`; rows where the predicate is `null` must be excluded. Grouped `first_value` / `last_value` checked only `BooleanArray::value(idx)`, without checking validity, so a NULL predicate row whose underlying value bit is set (as produced by comparison kernels, e.g. `null::int < 1`) was treated as passing: ```sql 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; -- returned fv = 10, must return fv = NULL -- (row 1: b < 1 is NULL; row 2: b < 1 is FALSE — no row satisfies `true`) ``` ## What changes are included in this PR? `datafusion/functions-aggregate/src/first_last.rs`, in `FirstLastGroupsAccumulator::get_filtered_extreme_of_each_group` (shared by `first_value` and `last_value`, and by both the `update_batch` and `merge_batch` paths): - `passed_filter` now requires `is_valid(idx) && value(idx)` (the `Some(true)` semantics), matching the convention already used by `variance.rs` / `correlation.rs`. - The `is_set_arr` read gets the same validity check. This is *not* only an internal bitmap: `convert_to_state` stores the user FILTER clause (including its nulls) in the last state column, so on the merge path (e.g. skip-partial-aggregation) NULL predicate rows were likewise treated as set. Verified with a forced skip-partial run (100k unique groups, all-NULL predicates): 83,616 groups were incorrectly assigned non-NULL values before the fix, 0 after. For genuine internal bitmaps (no nulls) the added check is trivially true, so behavior there is unchanged. Regression coverage: - sqllogictest (`aggregate.slt`): the issue reproducer, the `last_value` counterpart, mixed TRUE/FALSE/NULL predicates, all-TRUE and no-FILTER controls, the (already correct) non-grouped path, and a window-function no-regression case. - Unit tests: `test_group_acc_filter_null_predicate` (update path) and `test_group_acc_merge_null_is_set` (merge path via `convert_to_state` → `merge_batch`), both constructing `BooleanArray`s whose null slots carry a set value bit. ## Are these changes tested? Yes — see above. Verified `./dev/rust_lint.sh`, `cargo test -p datafusion-functions-aggregate --lib`, the `aggregate`/`window` sqllogictest files, and `datafusion-cli` end-to-end (grouped first/last_value now return NULL for the issue reproducer; mixed-predicate and non-grouped results unchanged). Also audited the rest of `functions-aggregate` for the same validity-blind pattern: shared helpers (`nulls.rs::filter_to_validity`, `accumulate.rs`, `prim_op.rs`, `count.rs`, `array_agg.rs`, `variance.rs`, `correlation.rs`) already handle validity correctly, and non-grouped paths pre-filter with arrow's `filter` kernel (which drops NULL predicate rows), so no other aggregate needs changes. ## Performance `functions-aggregate/benches/first_last.rs` was run against `main`. The added checks are one validity-bit test per row on the grouped path; measured deltas were within the machine's noise floor (±5% on unchanged `filter=false` cases). A variant hoisting the null check out of the row loop showed no measurable benefit beyond noise, so the simple idiomatic form is kept. ## Are there any user-facing changes? Only the bug fix: grouped `first_value`/`last_value` with a nullable `FILTER` predicate now correctly exclude NULL-predicate rows, matching SQL semantics and the behavior of other aggregates. No API or configuration changes. --------- Co-authored-by: Claude <[email protected]> --- datafusion/functions-aggregate/src/first_last.rs | 124 ++++++++++++++++++++++- datafusion/sqllogictest/test_files/aggregate.slt | 63 ++++++++++++ 2 files changed, 185 insertions(+), 2 deletions(-) diff --git a/datafusion/functions-aggregate/src/first_last.rs b/datafusion/functions-aggregate/src/first_last.rs index cecb277cb8..7c2540aa03 100644 --- a/datafusion/functions-aggregate/src/first_last.rs +++ b/datafusion/functions-aggregate/src/first_last.rs @@ -555,8 +555,15 @@ impl<S: ValueState> FirstLastGroupsAccumulator<S> { 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::<Int64Type>::new(DataType::Int64), + sort_keys.into(), + true, + &[DataType::Int64], + true, + )?; + + let val_with_orderings: Vec<ArrayRef> = 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::<Int64Array>().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<Int64Type> = 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::<Int64Type>::new(DataType::Int64), + sort_keys.clone().into(), + true, + &[DataType::Int64], + true, + )?; + + let val_with_orderings: Vec<ArrayRef> = 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::<Int64Type>::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::<Int64Array>().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<Int64Type> = 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<usize> { diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt index 9400a09a5d..1515e17e3f 100644 --- a/datafusion/sqllogictest/test_files/aggregate.slt +++ b/datafusion/sqllogictest/test_files/aggregate.slt @@ -6689,6 +6689,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) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
