perf: filter only aggregate argument columns by xudong963 · Pull Request #27119 · apache/datafusion

xudong963 · 2026-11-05T03:26:45Z

Which issue does this PR close?

Closes Aggregates with FILTER (WHERE ...) copy every input column once per aggregate #27114.

Rationale for this change

Each aggregate with FILTER currently filters every input column before evaluating its arguments. Queries with several filtered aggregates repeatedly copy columns that each aggregate does not read. This change filters only the columns needed by that aggregate, while preserving evaluation after filtering so rejected rows cannot cause errors such as division by zero.

What changes are included in this PR?

Record the input columns used by each aggregate's arguments, including its ORDER BY expressions. Unread columns use data-free NullArray placeholders to keep column indices valid.
Apply this to ungrouped aggregation, compacting grouped hash aggregation, and skip-partial state conversion. Skip-partial conversion shares one selected batch across all arguments.
Add SQL regressions for multiple aggregates, ordering, NULL and empty selections, fallible arguments, multiple partitions, and multi-argument skip-partial conversion. Physical-plan tests verify that unused columns are not copied and that skip-partial execution is exercised.

For the issue's 20M-row reproducer with 15 filtered aggregates, local medians of 21 warm runs after three warm-ups were:

Query	main (`70f2ed01b`)	This change	Speedup
No `GROUP BY`	68 ms	37 ms	1.84x
`GROUP BY quantity % 8`	95 ms	56 ms	1.70x

Both builds used release-nonlto; all results matched. The corresponding CASE WHEN controls measured 46/47 ms and 80/86 ms. Other local builds were running concurrently, so these timings include scheduling noise.

What is the testing strategy for this PR?

cargo fmt --all and cargo clippy --profile ci --all-targets --all-features -- -D warnings passed.
Ran 10 representative cases from the existing aggregate_filter Criterion benchmark against main, covering column/literal arguments, selection rates, group cardinality, and unfiltered controls. An initial 5.8% slowdown in an unfiltered control did not consistently reproduce in three alternating-order repetitions (fix/main ratios 0.998, 0.943, and 1.019).

Are there any user-facing changes?

Lower copying overhead for aggregate FILTER queries. SQL behavior and public APIs are unchanged.

Filter only the input columns read by aggregate arguments and ORDER BY expressions. Share the selected batch across skip-partial arguments while preserving FILTER-before-evaluation semantics.

adriangb

Thanks! Some small nits that probably meed performance confirmation.

adriangb · 2026-11-09T16:00:16Z

+        batch: &RecordBatch,
+        selection: &BooleanArray,
+    ) -> Result<RecordBatch> {
+        let schema = batch.schema_ref();


Nit: skip-partial calls select for every batch, also when the filter keeps all rows or no rows. evaluate_compacted_args skips these cases. A null rejects its row, and has_false does not count nulls, so the all-rows check also needs null_count() == 0.

Suggested change

let schema = batch.schema_ref();

// A null in `selection` rejects its row, like `false`.

if !selection.has_true() {

return Ok(batch.slice(0, 0));

}

if selection.null_count() == 0 && !selection.has_false() {

return Ok(batch.clone());

}

let schema = batch.schema_ref();

With an always-true filter on the #27114 data and a high-cardinality GROUP BY, the summed arguments_time is 15 ms on main, 31 ms on this PR, and 19 ms with this change. Wall time does not change.

Added the all-selected fast path with the null_count() == 0 guard. For empty selections, the skip-partial caller now directly creates typed null arguments and retains the original filter, avoiding selected-batch construction entirely. I avoided batch.slice(0, 0) because it still slices every input column, which was slower for wide batches in local testing. In a local all-selected/high-cardinality run, summed partial arguments_time dropped from about 12.6 ms to 6.4 ms, with wall time essentially unchanged.

adriangb · 2026-11-09T16:00:16Z

-use arrow::array::{ArrayRef, BooleanArray};
-use arrow::record_batch::RecordBatch;
-use datafusion_common::{Result, ScalarValue};
+use arrow::array::{ArrayRef, BooleanArray, NullArray, new_null_array};


Needed for null_count.

Suggested change

use arrow::array::{ArrayRef, BooleanArray, NullArray, new_null_array};

use arrow::array::{Array, ArrayRef, BooleanArray, NullArray, new_null_array};

Done, added the Array import for null_count().

adriangb · 2026-11-09T16:00:16Z

+        let next = filter.select(&batch, &BooleanArray::from(vec![true; 4]))?;
+        assert_eq!(next.num_rows(), 4);


With the fast path, an all-true selection returns the input schema. A partial selection keeps this check.

Suggested change

let next = filter.select(&batch, &BooleanArray::from(vec![true; 4]))?;

assert_eq!(next.num_rows(), 4);

let next =

filter.select(&batch, &BooleanArray::from(vec![true, true, false, true]))?;

assert_eq!(next.num_rows(), 3);

Updated the cache check to use two partial selections. I also added separate assertions that an all-selected result shares the original schema and each input array.

alamb

Thanks @xudong963 and @adriangb -- this makes sense to me

alamb · 2026-11-09T20:41:09Z

+        if selected.num_rows() == 0 {
+            // A fallible expression may fail even for an empty batch, so it
+            // is not evaluated.
+            let data_type = self.expr.data_type(batch.schema_ref())?;


The group by should already know its output -- calling data_type once per batch is non trivial in some cases -- I woudl recommend computing this once and then just copying the type per batch

Done. AggregateArgument now computes and stores its expression’s type during initialization, using the original input schema.

Empty-selection paths reuse that type, and accumulator reconstruction preserves it. This also handles cases such as AVG, where the argument and aggregate result types differ.

alamb · 2026-11-09T20:42:23Z

+        let selection = self
+            .predicate
+            .evaluate(batch)?
+            .into_array(batch.num_rows())?;


why turn it into an array? If it was a ColumnarValue::Scalar we could probably have a special path (can do as a follow on)

Agreed, a scalar predicate could avoid materializing a BooleanArray. I'll leave this for a follow-up it would need to handle scalar true/false/NULL consistently across the compacted and skip-partial paths.

alamb · 2026-11-09T20:43:40Z

+        let schema = batch.schema_ref();
+        // As in `filter_record_batch`, optimizing the filter pays off only
+        // when it is applied to several arrays.
+        let optimize = match self.argument_columns.as_slice() {


how about just using https://docs.rs/arrow/latest/arrow/array/struct.RecordBatch.html#method.project

and then calling the actual filter kernel?

I think that would be a lot faster and you can take advantage of all the crazy filter ing tricks in arrow

The current implementation already usesArrow’s FilterBuilderand FilterPredicate::filter. The complication with RecordBatch::project is that it changes column positions, while the argument expressions still reference the original indices.

I tried projecting, filtering, and restoring the original positions locally, but that was slightly slower in the cases tested. I’ve kept the current approach here; remapping the expressions once to use a compact projected batch could be worth exploring separately.

alamb · 2026-11-09T20:44:58Z

+            return Arc::clone(selected);
        }
+
+        let fields = schema


This looks like maybe it is a manual reimplementation of https://docs.rs/arrow/latest/arrow/datatypes/struct.Schema.html#method.project ?

Schema::project removes fields and changes their positions. This helper needs to preserve the original positions and replace unused fields with Null so existing argument expressions remain valid. I added a comment explaining why projection is not a direct replacement here.

alamb · 2026-11-09T20:45:28Z

    #[test]
-    fn literal_selection_matches_evaluate_selection() -> Result<()> {
+    fn selection_matches_physical_expr_evaluate_selection() -> Result<()> {
+        // a = [0, 1, 2]


can we test this with SLT rather than with a new unit test?

done, nice suggestion!

alamb · 2026-11-09T20:46:49Z


+    /// Aggregate argument that reads `column` and records the data types of
+    /// the columns of every batch it is evaluated against.
+    #[derive(Debug)]


this seems like a lot of test code, but it does test a reasonable property so it is probably ok

Thanks, I kept this property test. It checks that unused columns are replaced with placeholders during argument evaluation, which SQL result tests cannot observe.

Reuse all-selected batches, bypass empty skip-partial selections, and cache aggregate argument types. Move selection semantics coverage to SQLLogicTest.

adriangb

thanks for the great suggestions @alamb. lgtm!

perf: filter only aggregate argument columns

7578e9f

Filter only the input columns read by aggregate arguments and ORDER BY expressions. Share the selected batch across skip-partial arguments while preserving FILTER-before-evaluation semantics.

github-actions Bot added sqllogictest SQL Logic Tests (.slt) physical-plan Changes to the physical-plan crate labels Nov 5, 2026

xudong963 marked this pull request as ready for review November 9, 2026 02:53

xudong963 mentioned this pull request Nov 9, 2026

Release DataFusion 56.0.0 (Oct/Nov 2026) #24461

Open

40 of 61 tasks

adriangb approved these changes Nov 9, 2026

View reviewed changes

alamb approved these changes Nov 9, 2026

View reviewed changes

perf: avoid redundant aggregate FILTER work

1d39c54

Reuse all-selected batches, bypass empty skip-partial selections, and cache aggregate argument types. Move selection semantics coverage to SQLLogicTest.

adriangb approved these changes Nov 10, 2026

View reviewed changes

Dandandan approved these changes Nov 10, 2026

View reviewed changes

adriangb added this pull request to the merge queue Nov 10, 2026

github-merge-queue Bot removed this pull request from the merge queue due to failed status checks Nov 10, 2026

xudong963 added this pull request to the merge queue Nov 11, 2026

github-merge-queue Bot removed this pull request from the merge queue due to failed status checks Nov 11, 2026

xudong963 added 2 commits November 11, 2026 09:07

Merge branch 'main' into xudong963/fix-27114-filter-columns

1ae0651

fix: restore AsArray import after merging main

7e8f6c8

xudong963 added this pull request to the merge queue Nov 11, 2026

Merged via the queue into apache:main with commit 34eb6dc Nov 11, 2026
44 checks passed

xudong963 deleted the xudong963/fix-27114-filter-columns branch November 11, 2026 02:40

github-actions Bot added the v57.0.0 label Nov 11, 2026

xudong963 mentioned this pull request Nov 11, 2026

perf: specialize scalar aggregate FILTER evaluation #27321

Open

-        let schema = batch.schema_ref();
+        // A null in `selection` rejects its row, like `false`.
+        if !selection.has_true() {
+            return Ok(batch.slice(0, 0));
+        }
+        if selection.null_count() == 0 && !selection.has_false() {
+            return Ok(batch.clone());
+        }
+        let schema = batch.schema_ref();

	use arrow::array::{ArrayRef, BooleanArray, NullArray, new_null_array};
	use arrow::array::{Array, ArrayRef, BooleanArray, NullArray, new_null_array};

		let next = filter.select(&batch, &BooleanArray::from(vec![true; 4]))?;
		assert_eq!(next.num_rows(), 4);

Conversation

xudong963 commented Nov 5, 2026 • edited by alamb Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

What is the testing strategy for this PR?

Are there any user-facing changes?

Uh oh!

adriangb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xudong963 Nov 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adriangb left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

xudong963 commented Nov 5, 2026 •

edited by alamb

Loading

xudong963 Nov 10, 2026 •

edited

Loading