You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/24 09:31:20 UTC
(datafusion) branch main updated: fix: grouped first_value/last_value FILTER excludes NULL predicate rows (#23707)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 18b1e359c3 fix: grouped first_value/last_value FILTER excludes NULL predicate rows (#23707)
18b1e359c3 is described below
commit 18b1e359c3c547ec0d649932f85abbe33144ef19
Author: kid <19...@users.noreply.github.com>
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 <no...@anthropic.com>
---
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org