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/10/01 12:16:56 UTC
(datafusion) branch gh-readonly-queue/main/pr-25723-5635e5c1965fcb8b52201ebdfc6b3748f25ff139 created (now 84b84e5700)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25723-5635e5c1965fcb8b52201ebdfc6b3748f25ff139
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 84b84e5700 perf: Optimize `array_agg(... IGNORE NULLS)` (#25723)
This branch includes the following new commits:
new 84b84e5700 perf: Optimize `array_agg(... IGNORE NULLS)` (#25723)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: perf: Optimize `array_agg(... IGNORE NULLS)` (#25723)
Posted by gi...@apache.org.
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-25723-5635e5c1965fcb8b52201ebdfc6b3748f25ff139
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 84b84e5700fd69c581382dc475fa072a7375faba
Author: Neil Conway <ne...@gmail.com>
AuthorDate: Thu Oct 1 12:16:28 2026 +0000
perf: Optimize `array_agg(... IGNORE NULLS)` (#25723)
## Which issue does this PR close?
- N/A
## Rationale for this change
When `IGNORE NULLS` is specified, `array_agg` discards NULLs from its
input by passing a "keep" bitmap to the `filter` kernel. The distinct
and ordered `array_agg` accumulators built this keep bitmap bit-by-bit,
by iterating over the input's logical null buffer. That is inefficient:
we can just use the null buffer as-is. The plain `array_agg` accumulator
already did just that.
Benchmarks: `distinct_array_agg` benchmark, M4 Max, 8192 rows
- low cardinality: unchanged
- high cardinality: unchanged
- low cardinality, 10% nulls, ignore nulls: 109.5 us -> 102.2 us (6.7%
faster)
- high cardinality, 10% nulls, ignore nulls: 244.0 us -> 236.0 us (3.3%
faster)
## What changes are included in this PR?
* Introduce a shared helper, `non_null_mask`, to return the keep mask
* Use the new helper in the distinct and ordered `array_agg`
accumulators
* Extend benchmark to cover `IGNORE NULLS`
## What is the testing strategy for this PR?
Covered by existing tests.
## Are there any user-facing changes?
No.
---
.../functions-aggregate/benches/array_agg.rs | 55 +++++++++++++++-----
datafusion/functions-aggregate/src/array_agg.rs | 58 ++++++++++------------
2 files changed, 69 insertions(+), 44 deletions(-)
diff --git a/datafusion/functions-aggregate/benches/array_agg.rs b/datafusion/functions-aggregate/benches/array_agg.rs
index ebfe1f8cdd..4fcb2b32f2 100644
--- a/datafusion/functions-aggregate/benches/array_agg.rs
+++ b/datafusion/functions-aggregate/benches/array_agg.rs
@@ -227,29 +227,40 @@ const DB_NAMES: &[&str] = &[
/// Low-cardinality: every row is drawn uniformly from `DB_NAMES` (~25 distinct
/// values across 8 192 rows). Exercises the hot duplicate path.
-fn create_string_array_low_cardinality(size: usize) -> StringArray {
+/// `null_density` is the fraction of rows that are null instead.
+fn create_string_array_low_cardinality(size: usize, null_density: f32) -> StringArray {
let mut rng = StdRng::seed_from_u64(42);
- StringArray::from_iter_values(
- (0..size).map(|_| DB_NAMES[rng.random_range(0..DB_NAMES.len())]),
- )
+ StringArray::from_iter((0..size).map(|_| {
+ if null_density > 0.0 && rng.random::<f32>() < null_density {
+ return None;
+ }
+ Some(DB_NAMES[rng.random_range(0..DB_NAMES.len())])
+ }))
}
/// High-cardinality: `db_name_pct` fraction of rows are drawn from `DB_NAMES`;
/// the rest are near-unique random hex strings ("id_XXXXXXXX").
/// With 8 192 rows and a 32-bit space the collision probability among the
/// random strings is < 1 %, giving ~7 800 distinct values in total.
-fn create_string_array_high_cardinality(size: usize, db_name_pct: f32) -> StringArray {
+/// `null_density` is the fraction of rows that are null instead.
+fn create_string_array_high_cardinality(
+ size: usize,
+ db_name_pct: f32,
+ null_density: f32,
+) -> StringArray {
let mut rng = StdRng::seed_from_u64(42);
- let strings: Vec<String> = (0..size)
+ let strings: Vec<Option<String>> = (0..size)
.map(|_| {
- if rng.random::<f32>() < db_name_pct {
- DB_NAMES[rng.random_range(0..DB_NAMES.len())].to_string()
+ if null_density > 0.0 && rng.random::<f32>() < null_density {
+ None
+ } else if rng.random::<f32>() < db_name_pct {
+ Some(DB_NAMES[rng.random_range(0..DB_NAMES.len())].to_string())
} else {
- format!("id_{:08x}", rng.random::<u32>())
+ Some(format!("id_{:08x}", rng.random::<u32>()))
}
})
.collect();
- StringArray::from_iter_values(strings.iter().map(String::as_str))
+ StringArray::from_iter(strings.iter().map(Option::as_deref))
}
fn distinct_update_batch_bench(
@@ -272,7 +283,7 @@ fn distinct_array_agg_benchmark(c: &mut Criterion) {
// --- Low cardinality: ~25 distinct DB names in 8 192 rows ---------------
// Realistic production scenario: most rows are duplicates, the HashSet
// saturates quickly and the rest of the batch is pure dedup overhead.
- let values = Arc::new(create_string_array_low_cardinality(8192)) as ArrayRef;
+ let values = Arc::new(create_string_array_low_cardinality(8192, 0.0)) as ArrayRef;
distinct_update_batch_bench(
c,
"distinct_array_agg utf8 low cardinality (~25 distinct)",
@@ -283,13 +294,33 @@ fn distinct_array_agg_benchmark(c: &mut Criterion) {
// --- High cardinality: ~5 % DB names, ~95 % near-unique random strings --
// Worst-case scenario: almost every row is a new distinct value, so the
// accumulator pays the full insertion cost for nearly every row.
- let values = Arc::new(create_string_array_high_cardinality(8192, 0.05)) as ArrayRef;
+ let values =
+ Arc::new(create_string_array_high_cardinality(8192, 0.05, 0.0)) as ArrayRef;
distinct_update_batch_bench(
c,
"distinct_array_agg utf8 high cardinality (~7800 distinct, 5% db names)",
&values,
false,
);
+
+ // --- With nulls and IGNORE NULLS: the accumulator drops the null rows
+ // before deduplicating, so these cover that filtering step.
+ let values = Arc::new(create_string_array_low_cardinality(8192, 0.1)) as ArrayRef;
+ distinct_update_batch_bench(
+ c,
+ "distinct_array_agg utf8 low cardinality (~25 distinct), 10% nulls, ignore nulls",
+ &values,
+ true,
+ );
+
+ let values =
+ Arc::new(create_string_array_high_cardinality(8192, 0.05, 0.1)) as ArrayRef;
+ distinct_update_batch_bench(
+ c,
+ "distinct_array_agg utf8 high cardinality (~7800 distinct, 5% db names), 10% nulls, ignore nulls",
+ &values,
+ true,
+ );
}
/// Precomputes the schema, physical expressions, sort expression, and aggregate
diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs
index abf1c8376f..70a04e0fe9 100644
--- a/datafusion/functions-aggregate/src/array_agg.rs
+++ b/datafusion/functions-aggregate/src/array_agg.rs
@@ -339,6 +339,20 @@ impl ArrayAggAccumulator {
}
}
+/// Returns the mask of non-null rows in `values` when `ignore_nulls` is set
+/// and some row is null.
+fn non_null_mask(values: &dyn Array, ignore_nulls: bool) -> Option<BooleanArray> {
+ if !ignore_nulls {
+ return None;
+ }
+ let nulls = values.logical_nulls()?;
+ if nulls.null_count() == 0 {
+ return None;
+ }
+ // The validity bitmap is the keep mask: a row is kept when it is valid.
+ Some(BooleanArray::new(nulls.into_inner(), None))
+}
+
impl Accumulator for ArrayAggAccumulator {
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
// Append value like Int64Array(1,2,3)
@@ -924,20 +938,12 @@ impl DistinctArrayAggAccumulator {
// Filter nulls out upfront when ignore_nulls is set so they are
// never inserted into the dedup state.
let filtered;
- let col: &ArrayRef = if self.ignore_nulls {
- if let Some(nulls) = val.logical_nulls() {
- if nulls.null_count() > 0 {
- let mask: BooleanArray = nulls.iter().map(Some).collect();
- filtered = filter(val.as_ref(), &mask)?;
- &filtered
- } else {
- val
- }
- } else {
- val
+ let col: &ArrayRef = match non_null_mask(val.as_ref(), self.ignore_nulls) {
+ Some(mask) => {
+ filtered = filter(val.as_ref(), &mask)?;
+ &filtered
}
- } else {
- val
+ None => val,
};
if col.is_empty() {
@@ -1122,20 +1128,12 @@ impl Accumulator for DistinctArrayAggAccumulator {
// Mirror the null-filtering logic from update_batch so we only
// retract values that were actually inserted.
let filtered;
- let col: &ArrayRef = if self.ignore_nulls {
- if let Some(nulls) = val.logical_nulls() {
- if nulls.null_count() > 0 {
- let mask: BooleanArray = nulls.iter().map(Some).collect();
- filtered = filter(val.as_ref(), &mask)?;
- &filtered
- } else {
- val
- }
- } else {
- val
+ let col: &ArrayRef = match non_null_mask(val.as_ref(), self.ignore_nulls) {
+ Some(mask) => {
+ filtered = filter(val.as_ref(), &mask)?;
+ &filtered
}
- } else {
- val
+ None => val,
};
if col.is_empty() {
@@ -1508,12 +1506,8 @@ impl OrderSensitiveArrayAggAccumulator {
);
}
- let nulls = ignore_nulls
- .then(|| values.logical_nulls())
- .flatten()
- .filter(|nulls| nulls.null_count() > 0);
- let (values, filtered_ordering_values) = if let Some(nulls) = nulls {
- let mask: BooleanArray = nulls.iter().map(Some).collect();
+ let mask = non_null_mask(values.as_ref(), ignore_nulls);
+ let (values, filtered_ordering_values) = if let Some(mask) = mask {
let values = filter(values.as_ref(), &mask)?;
let ordering_values = ordering_values
.iter()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org