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/09/17 17:54:30 UTC
(datafusion) branch main updated: Avoid normalizing hidden values in sliced list set operations (#25300)
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 bf67e970ba Avoid normalizing hidden values in sliced list set operations (#25300)
bf67e970ba is described below
commit bf67e970ba957df224917105e9a0bc8d98cc3f0b
Author: Yin Li <kx...@student.bham.ac.uk>
AuthorDate: Thu Sep 17 17:45:13 2026 +0000
Avoid normalizing hidden values in sliced list set operations (#25300)
## Which issue does this PR close?
- Related to https://github.com/apache/datafusion-comet/issues/5708
(tracked under https://github.com/apache/datafusion-comet/issues/5709).
## Rationale for this change
A sliced `ListArray` can retain a child values array that is much larger
than its logical range. The floating-point set-operation paths currently
call `normalize_float_zero()` on that full child array before slicing it
to the visible offsets. When the backing array contains `-0.0`, this
scans and allocates in proportion to hidden backing data rather than the
values the query can observe.
This is particularly expensive for small slices of large arrays. In the
added Criterion benchmark, `array_distinct` over two visible values
backed by 1,048,576 `Float64` values takes 184.46 us on the parent
commit and 0.823 us with this change (about 224x faster). This is a
component benchmark, not an end-to-end query speedup.
## What changes are included in this PR?
- Slice each list's visible child range before floating-point zero
normalization in `array_distinct`, `array_union`, `array_intersect`, and
`array_except`.
- Keep row and output indices relative to the normalized slice,
including the right-hand concatenation offset used by union and
intersection.
- Preserve the existing full-child path when the entire child array is
visible.
- Add a benchmark that independently varies backing-buffer size and
visible-range size, plus an unsliced control.
- Add regression coverage for non-zero offsets, `List` and `LargeList`,
signed zero, repeated NaNs, null elements, null list rows, and an empty
visible list.
Benchmark results on the same machine and parent commit (`c149764`),
using Criterion's 100-sample estimates:
| Input | Parent | This PR |
|---|---:|---:|
| 2 visible values, 1,024 backing values | 1.024 us | 0.858 us |
| 2 visible values, 1,048,576 backing values | 184.46 us | 0.823 us |
| 2,048 visible values, 1,048,576 backing values | 167.27 us | 12.50 us
|
With two visible values, the new implementation remains approximately
flat as the backing array grows from 1,024 to 1,048,576 values.
Increasing the visible range still increases runtime as expected.
## What is the testing strategy for this PR?
The new unit tests exercise all affected set operations and make the
pre-slice and in-slice values intentionally different so incorrect
absolute/local index conversion is observable.
Validation completed locally:
- `cargo test -p datafusion-functions-nested` (127 tests and 2 doctests)
- Extended workspace test command from `AGENTS.md`, including all 519
sqllogictest files
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo fmt --all -- --check`
- `cargo bench -p datafusion-functions-nested --bench array_set_ops --
array_distinct_sliced_float`
## Are there any user-facing changes?
There are no API or result-semantics changes. Sliced floating-point list
set operations avoid work and allocation for child values outside their
logical range while retaining the existing signed-zero, NaN, and null
behavior.
---
.../functions-nested/benches/array_set_ops.rs | 105 ++++++------
datafusion/functions-nested/src/except.rs | 57 ++++---
datafusion/functions-nested/src/set_ops.rs | 176 +++++++++++++++++----
3 files changed, 245 insertions(+), 93 deletions(-)
diff --git a/datafusion/functions-nested/benches/array_set_ops.rs b/datafusion/functions-nested/benches/array_set_ops.rs
index d43bbdb577..9c51974cd3 100644
--- a/datafusion/functions-nested/benches/array_set_ops.rs
+++ b/datafusion/functions-nested/benches/array_set_ops.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use arrow::array::{ArrayRef, Int64Array, ListArray};
+use arrow::array::{ArrayRef, Float64Array, Int64Array, ListArray};
use arrow::buffer::OffsetBuffer;
use arrow::datatypes::{DataType, Field};
use criterion::{
@@ -47,6 +47,7 @@ fn criterion_benchmark(c: &mut Criterion) {
bench_array_union_sliced(c);
bench_array_intersect_sliced(c);
bench_array_distinct_sliced(c);
+ bench_array_distinct_sliced_float(c);
bench_array_except_sliced(c);
}
@@ -69,6 +70,23 @@ fn invoke_udf(udf: &impl ScalarUDFImpl, array1: &ArrayRef, array2: &ArrayRef) {
);
}
+fn invoke_unary_udf(
+ udf: &impl ScalarUDFImpl,
+ array: &ArrayRef,
+ number_rows: usize,
+) -> ColumnarValue {
+ black_box(
+ udf.invoke_with_args(ScalarFunctionArgs {
+ args: vec![ColumnarValue::Array(array.clone())],
+ arg_fields: vec![Field::new("arr", array.data_type().clone(), false).into()],
+ number_rows,
+ return_field: Field::new("result", array.data_type().clone(), false).into(),
+ config_options: Arc::new(ConfigOptions::default()),
+ })
+ .unwrap(),
+ )
+}
+
fn bench_array_union(c: &mut Criterion) {
let mut group = c.benchmark_group("array_union");
let udf = ArrayUnion::new();
@@ -139,28 +157,7 @@ fn bench_array_distinct(c: &mut Criterion) {
group.bench_with_input(
BenchmarkId::new(*duplicate_label, array_size),
&array_size,
- |b, _| {
- b.iter(|| {
- black_box(
- udf.invoke_with_args(ScalarFunctionArgs {
- args: vec![ColumnarValue::Array(array.clone())],
- arg_fields: vec![
- Field::new("arr", array.data_type().clone(), false)
- .into(),
- ],
- number_rows: NUM_ROWS,
- return_field: Field::new(
- "result",
- array.data_type().clone(),
- false,
- )
- .into(),
- config_options: Arc::new(ConfigOptions::default()),
- })
- .unwrap(),
- )
- })
- },
+ |b, _| b.iter(|| invoke_unary_udf(&udf, &array, NUM_ROWS)),
);
}
}
@@ -358,32 +355,50 @@ fn bench_array_distinct_sliced(c: &mut Criterion) {
group.bench_with_input(
BenchmarkId::from_parameter(array_size),
&array_size,
- |b, _| {
- b.iter(|| {
- black_box(
- udf.invoke_with_args(ScalarFunctionArgs {
- args: vec![ColumnarValue::Array(array.clone())],
- arg_fields: vec![
- Field::new("arr", array.data_type().clone(), false)
- .into(),
- ],
- number_rows: NUM_ROWS,
- return_field: Field::new(
- "result",
- array.data_type().clone(),
- false,
- )
- .into(),
- config_options: Arc::new(ConfigOptions::default()),
- })
- .unwrap(),
- )
- })
- },
+ |b, _| b.iter(|| invoke_unary_udf(&udf, &array, NUM_ROWS)),
);
}
group.finish();
}
+fn create_sliced_float_array(backing_values: usize, visible_values: usize) -> ArrayRef {
+ assert!(visible_values > 0 && visible_values <= backing_values);
+
+ let values = Float64Array::from(
+ (0..backing_values)
+ .map(|i| if i.is_multiple_of(2) { -0.0 } else { 0.0 })
+ .collect::<Vec<_>>(),
+ );
+ let left_padding = (backing_values - visible_values) / 2;
+ let offsets = vec![
+ 0,
+ left_padding as i32,
+ (left_padding + visible_values) as i32,
+ backing_values as i32,
+ ];
+ let array = ListArray::try_new(
+ Arc::new(Field::new("item", DataType::Float64, true)),
+ OffsetBuffer::new(offsets.into()),
+ Arc::new(values),
+ None,
+ )
+ .unwrap();
+
+ Arc::new(array.slice(1, 1))
+}
+
+/// Keep the visible list fixed at one `-0.0` and one `0.0` inside a much larger
+/// backing values buffer to catch regressions that process values outside the slice.
+fn bench_array_distinct_sliced_float(c: &mut Criterion) {
+ let mut group = c.benchmark_group("array_distinct_sliced_float");
+ let udf = ArrayDistinct::new();
+ let array = create_sliced_float_array(1024 * 1024, 2);
+
+ group.bench_function("backing_values_1048576_visible_values_2", |b| {
+ b.iter(|| invoke_unary_udf(&udf, &array, 1))
+ });
+ group.finish();
+}
+
criterion_group!(benches, criterion_benchmark);
criterion_main!(benches);
diff --git a/datafusion/functions-nested/src/except.rs b/datafusion/functions-nested/src/except.rs
index 737a3122bb..fd2b6a0897 100644
--- a/datafusion/functions-nested/src/except.rs
+++ b/datafusion/functions-nested/src/except.rs
@@ -17,6 +17,7 @@
//! [`ScalarUDFImpl`] definition for array_except function.
+use crate::set_ops::normalize_visible_values;
use crate::utils::{check_datatypes, make_scalar_function};
use arrow::array::new_null_array;
use arrow::array::{
@@ -27,7 +28,7 @@ use arrow::buffer::{NullBuffer, OffsetBuffer};
use arrow::compute::take;
use arrow::datatypes::{DataType, FieldRef};
use arrow::row::{RowConverter, SortField};
-use datafusion_common::utils::{ListCoercion, normalize_float_zero, take_function_args};
+use datafusion_common::utils::{ListCoercion, take_function_args};
use datafusion_common::{HashSet, Result, internal_err};
use datafusion_expr::{
ColumnarValue, Documentation, ScalarFunctionArgs, ScalarUDFImpl, Signature,
@@ -171,24 +172,18 @@ fn general_except<OffsetSize: OffsetSizeTrait>(
// Normalize -0.0 → +0.0 so RowConverter (IEEE 754 totalOrder) groups
// ±0 together for both the rhs lookup set and the lhs probe.
- let l_values_norm = normalize_float_zero(l.values());
- let r_values_norm = normalize_float_zero(r.values());
-
- // Only convert the visible portion of the values array. For sliced
- // ListArrays, values() returns the full underlying array but only
- // elements between the first and last offset are referenced.
let l_first = l.offsets()[0].as_usize();
- let l_len = l.offsets()[l.len()].as_usize() - l_first;
- let l_values = converter.convert_columns(&[l_values_norm.slice(l_first, l_len)])?;
+ let l_values_norm = normalize_visible_values(l);
+ let l_rows = converter.convert_columns(&[Arc::clone(&l_values_norm)])?;
let r_first = r.offsets()[0].as_usize();
- let r_len = r.offsets()[r.len()].as_usize() - r_first;
- let r_values = converter.convert_columns(&[r_values_norm.slice(r_first, r_len)])?;
+ let r_values_norm = normalize_visible_values(r);
+ let r_rows = converter.convert_columns(&[Arc::clone(&r_values_norm)])?;
let mut offsets = Vec::<OffsetSize>::with_capacity(l.len() + 1);
offsets.push(OffsetSize::usize_as(0));
- let mut indices: Vec<usize> = Vec::with_capacity(l_values.num_rows());
+ let mut indices: Vec<usize> = Vec::with_capacity(l_rows.num_rows());
let mut dedup = HashSet::new();
let nulls = NullBuffer::union(l.nulls(), r.nulls());
@@ -207,13 +202,13 @@ fn general_except<OffsetSize: OffsetSizeTrait>(
}
for element_index in r_start.as_usize() - r_first..r_end.as_usize() - r_first {
- let right_row = r_values.row(element_index);
+ let right_row = r_rows.row(element_index);
dedup.insert(right_row);
}
for element_index in l_start.as_usize() - l_first..l_end.as_usize() - l_first {
- let left_row = l_values.row(element_index);
+ let left_row = l_rows.row(element_index);
if dedup.insert(left_row) {
- indices.push(element_index + l_first);
+ indices.push(element_index);
}
}
@@ -245,9 +240,9 @@ fn general_except<OffsetSize: OffsetSizeTrait>(
#[cfg(test)]
mod tests {
- use super::ArrayExcept;
- use arrow::array::{Array, AsArray, Int32Array, ListArray};
- use arrow::datatypes::{Field, Int32Type};
+ use super::{ArrayExcept, general_except};
+ use arrow::array::{Array, AsArray, Int32Array, LargeListArray, ListArray};
+ use arrow::datatypes::{DataType, Field, Float64Type, Int32Type};
use datafusion_common::{Result, config::ConfigOptions};
use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl};
use std::sync::Arc;
@@ -302,4 +297,30 @@ mod tests {
Ok(())
}
+
+ #[test]
+ fn test_array_except_sliced_float_large_lists() -> Result<()> {
+ let l = LargeListArray::from_iter_primitive::<Float64Type, _, _>(vec![
+ Some(vec![Some(99.0)]),
+ Some(vec![Some(-0.0), Some(0.0), Some(1.0), None, None]),
+ Some(vec![Some(-99.0)]),
+ ])
+ .slice(1, 1);
+ let r = LargeListArray::from_iter_primitive::<Float64Type, _, _>(vec![
+ Some(vec![Some(98.0)]),
+ Some(vec![Some(0.0), None]),
+ Some(vec![Some(-98.0)]),
+ ])
+ .slice(1, 1);
+ let DataType::LargeList(field) = l.data_type() else {
+ unreachable!()
+ };
+
+ let result = general_except::<i64>(&l, &r, field)?;
+ let values = result.value(0);
+ let values = values.as_primitive::<Float64Type>();
+ assert_eq!(values.len(), 1);
+ assert_eq!(values.value(0), 1.0);
+ Ok(())
+ }
}
diff --git a/datafusion/functions-nested/src/set_ops.rs b/datafusion/functions-nested/src/set_ops.rs
index 2214d3d35b..5cd4819cae 100644
--- a/datafusion/functions-nested/src/set_ops.rs
+++ b/datafusion/functions-nested/src/set_ops.rs
@@ -329,6 +329,18 @@ impl Display for SetOp {
}
}
+pub(crate) fn normalize_visible_values<OffsetSize: OffsetSizeTrait>(
+ array: &GenericListArray<OffsetSize>,
+) -> ArrayRef {
+ let first = array.offsets()[0].as_usize();
+ let len = array.offsets()[array.len()].as_usize() - first;
+ if first == 0 && len == array.values().len() {
+ normalize_float_zero(array.values())
+ } else {
+ normalize_float_zero(&array.values().slice(first, len))
+ }
+}
+
fn generic_set_lists<OffsetSize: OffsetSizeTrait>(
l: &GenericListArray<OffsetSize>,
r: &GenericListArray<OffsetSize>,
@@ -354,27 +366,17 @@ fn generic_set_lists<OffsetSize: OffsetSizeTrait>(
// Normalize -0.0 → +0.0 so RowConverter (which uses IEEE 754 totalOrder
// and treats ±0 as distinct) groups them together. Use the normalized
// arrays for both row conversion and the final output values.
- let l_values_norm = normalize_float_zero(l.values());
- let r_values_norm = normalize_float_zero(r.values());
-
- // Only convert the visible portion of the values array. For sliced
- // ListArrays, values() returns the full underlying array but only
- // elements between the first and last offset are referenced.
- let l_first = l.offsets()[0].as_usize();
- let l_len = l.offsets()[l.len()].as_usize() - l_first;
- let l_values = l_values_norm.slice(l_first, l_len);
- let rows_l = converter.convert_columns(&[Arc::clone(&l_values)])?;
+ let l_values_norm = normalize_visible_values(l);
+ let r_values_norm = normalize_visible_values(r);
- let r_first = r.offsets()[0].as_usize();
- let r_len = r.offsets()[r.len()].as_usize() - r_first;
- let r_values = r_values_norm.slice(r_first, r_len);
- let rows_r = converter.convert_columns(&[Arc::clone(&r_values)])?;
+ let rows_l = converter.convert_columns(&[Arc::clone(&l_values_norm)])?;
+ let rows_r = converter.convert_columns(&[Arc::clone(&r_values_norm)])?;
// Indices from the row converter are 0-based in the per-side slice;
// concatenating those same slices lets indices map directly into the
// combined values array.
- let combined_values = concat(&[l_values.as_ref(), r_values.as_ref()])?;
- let r_offset = l_len;
+ let combined_values = concat(&[l_values_norm.as_ref(), r_values_norm.as_ref()])?;
+ let r_offset = l_values_norm.len();
match set_op {
SetOp::Union => generic_set_loop::<OffsetSize, true>(
@@ -568,15 +570,9 @@ fn general_array_distinct<OffsetSize: OffsetSizeTrait>(
// Normalize -0.0 → +0.0 so RowConverter (which uses IEEE 754 totalOrder
// and treats ±0 as distinct) groups them together, and so the output
// carries the canonical sign.
- let values_norm = normalize_float_zero(array.values());
-
- // Only convert the visible portion of the values array. For sliced
- // ListArrays, values() returns the full underlying array but only
- // elements between the first and last offset are referenced.
let first_offset = value_offsets[0].as_usize();
- let visible_len = value_offsets[array.len()].as_usize() - first_offset;
- let rows =
- converter.convert_columns(&[values_norm.slice(first_offset, visible_len)])?;
+ let values_norm = normalize_visible_values(array);
+ let rows = converter.convert_columns(&[Arc::clone(&values_norm)])?;
let mut indices: Vec<usize> = Vec::with_capacity(rows.num_rows());
let mut seen = HashSet::new();
@@ -598,15 +594,14 @@ fn general_array_distinct<OffsetSize: OffsetSizeTrait>(
for idx in start..end {
let row = rows.row(idx);
if seen.insert(row) {
- indices.push(idx + first_offset);
+ indices.push(idx);
}
}
offsets.push(last_offset + OffsetSize::usize_as(seen.len()));
}
// Gather distinct values in a single pass, using the computed `indices`.
- // Indices are absolute positions in the (normalized) values array, so we
- // can take directly from the full values.
+ // Indices are relative to the visible, normalized values array.
// Use UInt64Array for LargeList to support values arrays exceeding u32::MAX.
let final_values = if indices.is_empty() {
new_empty_array(&dt)
@@ -634,14 +629,17 @@ mod tests {
use std::sync::Arc;
use arrow::{
- array::{Array, AsArray, Int32Array, ListArray},
+ array::{Array, AsArray, Int32Array, LargeListArray, ListArray},
buffer::OffsetBuffer,
- datatypes::{DataType, Field, Int32Type},
+ datatypes::{DataType, Field, Float64Type, Int32Type},
};
use datafusion_common::{DataFusionError, Result, config::ConfigOptions};
use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl};
- use crate::set_ops::{ArrayDistinct, ArrayIntersect, ArrayUnion, array_distinct_udf};
+ use crate::set_ops::{
+ ArrayDistinct, ArrayIntersect, ArrayUnion, SetOp, array_distinct_udf,
+ general_array_distinct, generic_set_lists,
+ };
/// Build two sliced ListArrays and return them along with the shared list
/// field.
@@ -678,6 +676,17 @@ mod tests {
.collect()
}
+ fn collect_f64_bits<OffsetSize: arrow::array::OffsetSizeTrait>(
+ list: &arrow::array::GenericListArray<OffsetSize>,
+ row: usize,
+ ) -> Vec<Option<u64>> {
+ list.value(row)
+ .as_primitive::<Float64Type>()
+ .iter()
+ .map(|value| value.map(f64::to_bits))
+ .collect()
+ }
+
#[test]
fn test_array_union_sliced_lists() -> Result<()> {
let (l, r, field) = make_sliced_pair();
@@ -761,6 +770,113 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_sliced_float_set_ops_preserve_semantics() -> Result<()> {
+ let nan = f64::from_bits(0x7ff8_0000_0000_0042);
+ let l = ListArray::from_iter_primitive::<Float64Type, _, _>(vec![
+ Some(vec![Some(99.0)]),
+ Some(vec![
+ Some(-0.0),
+ Some(0.0),
+ Some(1.0),
+ Some(nan),
+ Some(nan),
+ None,
+ None,
+ ]),
+ None,
+ Some(vec![Some(-99.0)]),
+ ])
+ .slice(1, 2);
+ let r = ListArray::from_iter_primitive::<Float64Type, _, _>(vec![
+ Some(vec![Some(98.0)]),
+ Some(vec![Some(0.0), Some(2.0), Some(nan), None]),
+ Some(vec![Some(1.0)]),
+ Some(vec![Some(-98.0)]),
+ ])
+ .slice(1, 2);
+ let DataType::List(field) = l.data_type() else {
+ unreachable!()
+ };
+
+ let distinct = general_array_distinct::<i32>(&l, field)?;
+ let distinct = distinct.as_list::<i32>();
+ assert_eq!(
+ collect_f64_bits(distinct, 0),
+ vec![
+ Some(0.0_f64.to_bits()),
+ Some(1.0_f64.to_bits()),
+ Some(nan.to_bits()),
+ None,
+ ]
+ );
+ assert!(distinct.is_null(1));
+
+ let union = generic_set_lists::<i32>(&l, &r, Arc::clone(field), SetOp::Union)?;
+ let union = union.as_list::<i32>();
+ assert_eq!(
+ collect_f64_bits(union, 0),
+ vec![
+ Some(0.0_f64.to_bits()),
+ Some(1.0_f64.to_bits()),
+ Some(nan.to_bits()),
+ None,
+ Some(2.0_f64.to_bits()),
+ ]
+ );
+ assert!(union.is_null(1));
+
+ let intersect =
+ generic_set_lists::<i32>(&l, &r, Arc::clone(field), SetOp::Intersect)?;
+ let intersect = intersect.as_list::<i32>();
+ assert_eq!(
+ collect_f64_bits(intersect, 0),
+ vec![Some(0.0_f64.to_bits()), Some(nan.to_bits()), None]
+ );
+ assert!(intersect.is_null(1));
+ Ok(())
+ }
+
+ #[test]
+ fn test_array_distinct_sliced_float_large_list() -> Result<()> {
+ let list = LargeListArray::from_iter_primitive::<Float64Type, _, _>(vec![
+ Some(vec![Some(99.0)]),
+ Some(vec![Some(-0.0), Some(0.0), Some(1.0), None, None]),
+ Some(vec![Some(-99.0)]),
+ ]);
+ let sliced = list.slice(1, 1);
+ let DataType::LargeList(field) = sliced.data_type() else {
+ unreachable!()
+ };
+
+ let result = general_array_distinct::<i64>(&sliced, field)?;
+ let result = result.as_list::<i64>();
+ assert_eq!(
+ collect_f64_bits(result, 0),
+ vec![Some(0.0_f64.to_bits()), Some(1.0_f64.to_bits()), None]
+ );
+ Ok(())
+ }
+
+ #[test]
+ fn test_array_distinct_sliced_empty_float_list() -> Result<()> {
+ let list = ListArray::from_iter_primitive::<Float64Type, _, _>(vec![
+ Some(vec![Some(-0.0), Some(99.0)]),
+ Some(Vec::<Option<f64>>::new()),
+ Some(vec![Some(-0.0), Some(0.0)]),
+ ]);
+ let sliced = list.slice(1, 1);
+ let DataType::List(field) = sliced.data_type() else {
+ unreachable!()
+ };
+
+ let result = general_array_distinct::<i32>(&sliced, field)?;
+ let result = result.as_list::<i32>();
+ assert_eq!(result.len(), 1);
+ assert_eq!(result.value_length(0), 0);
+ Ok(())
+ }
+
#[test]
fn test_array_distinct_inner_nullability_result_type_match_return_type()
-> Result<(), DataFusionError> {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org