Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions docs/source/contributor-guide/expression-audits/array_funcs.md
Original file line number Diff line number Diff line change
Expand Up @@ -122,9 +122,11 @@
## array_remove

- Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
- Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayRemove(left, right)`; removes all occurrences equal to `right`. Wired as `CometScalarFunction("array_remove")`. Falls back via `ArraysBase.isTypeSupported` for binary/struct/map/null child types.
- Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayRemove(left, right)`; removes all occurrences equal to `right`. Falls back via `ArraysBase.isTypeSupported` for binary/struct/map/null child types.
- Spark 4.0.1 (audited 2026-05-27): `NullIntolerant` -> `nullIntolerant` field refactor.
- Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
- Current status: `CometArrayRemove` sends arrays whose elements hold a `FLOAT` or `DOUBLE` at any depth to the native `spark_array_remove`, which compares as Spark's `genEqual` does (`-0.0` equals `0.0`, all NaNs are equal, nested elements through `spark_equality`) and keeps the bits of the elements it keeps. Other element types use DataFusion's `array_remove_all`. Null elements stay, and a null array or value gives null.
- Performance (tuned 2026-10-01, PR [#6518](https://github.com/apache/datafusion-comet/pull/6518)): with a constant value, `FLOAT` and `DOUBLE` elements get one keep mask over all the values from `BooleanBuffer::collect_bool`, with the inverted validity ORed in, one running popcount over the mask's words for each row's kept count, and one `filter`. 41-76% less time than DataFusion's `array_remove_all` with a constant value, and 47-95% less with a value per row. Benchmark: `benches/float_arrays.rs`.

## array_repeat

Expand Down Expand Up @@ -196,8 +198,10 @@
## sort_array

- Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
- Spark 3.5.8 (audited 2026-05-27): baseline. `SortArray(base, ascendingOrder) extends BinaryExpression with ArraySortLike`; the second arg must be a `Literal(_: Boolean, BooleanType)`. Comet `CometSortArray` flags `Incompatible` under strict floating-point and falls back for nested arrays whose innermost element is `Struct` or `Null`.
- Spark 4.0.1 (audited 2026-05-27): trait set changes substantively: `ArraySortLike` and `NullIntolerant` are removed, `nullIntolerant = true` becomes an override, and `ascendingOrder` is widened to accept any foldable boolean (not just `Literal`). Comet's `CometSortArray` still requires a `Literal`, so the new foldable form falls back at convert time.
- Spark 3.5.8 (audited 2026-05-27): baseline. `SortArray(base, ascendingOrder) extends BinaryExpression with ArraySortLike`; the second arg must be a `Literal(_: Boolean, BooleanType)`.
- Spark 4.0.1 (audited 2026-05-27): trait set changes substantively: `ArraySortLike` and `NullIntolerant` are removed, `nullIntolerant = true` becomes an override, and `ascendingOrder` is widened to accept any foldable boolean (not just `Literal`).
- Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
- Current status: elements with a `FLOAT` or `DOUBLE` at any depth go to the native `spark_sort_array`, which sorts as Spark's generated code does: a stable sort in Spark's SQL ordering, except that for an ascending sort of `FLOAT` or `DOUBLE` elements that cannot be null, the serde has it put `-0.0` before `0.0`, as `java.util.Arrays.sort` does. This holds in strict floating-point mode too. Other supported element types use DataFusion's `array_sort`, and unsupported ones route through the codegen dispatcher. `ascendingOrder` may be any foldable boolean, which the serde evaluates.
- Performance (tuned 2026-10-01, PR [#6518](https://github.com/apache/datafusion-comet/pull/6518)): `FLOAT` and `DOUBLE` rows sort within one copy of all the values, each row's valid values copied without branching next to its nulls. Rows of up to 20 elements use the stable `sort_by`. Longer rows use `sort_unstable_by` and then put the zero and NaN runs back in their original order, because Rust's stable sort is up to 1.8 times slower between 33 and 63 elements. From 2% more to 21% less time than DataFusion's `array_sort`. Benchmark: `benches/float_arrays.rs`.

[Spark Expression Support]: ../../user-guide/latest/expressions.md
12 changes: 12 additions & 0 deletions docs/source/user-guide/latest/compatibility/floating-point.md
Original file line number Diff line number Diff line change
Expand Up @@ -69,3 +69,15 @@ greater than non-NaN values. The original first equal element is retained: for e
`array_min(array(0.0D, -0.0D))` returns `0.0`, while reversing those elements returns `-0.0`.
The same ordering applies recursively to floating-point fields in arrays and structs. These
expressions do not require Spark's codegen dispatcher for floating-point compatibility.

## `array_remove` and `sort_array`

`array_remove` compares `FLOAT` and `DOUBLE` elements as Spark does: `-0.0` equals `0.0`, and all
NaN representations are equal, inside nested arrays too. The elements it keeps retain their
original NaN representations and zero signs.

`sort_array` sorts in Spark's order, in which NaN sorts above every other value and all NaN
representations tie, and keeps equal elements in their original order. In an array whose elements
can be null, such as one built from nullable columns, `-0.0` and `0.0` therefore tie. Sorting an
array whose elements cannot be null in ascending order, Spark's generated code puts `-0.0` before
`0.0`, and so does Comet. Both expressions run natively in strict floating-point mode.
2 changes: 1 addition & 1 deletion docs/source/user-guide/latest/expressions.md
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,7 @@ The tables below list every Spark built-in expression with its current status.
| `sequence` | ✅ | Hybrid | Integral types run natively; date/timestamp sequences use codegen dispatch |
| `shuffle` | ✅ | Native | Binary/struct/map elements fall back |
| `slice` | ✅ | Native | Native ([#4149](https://github.com/apache/datafusion-comet/pull/4149)) |
| `sort_array` | ✅ | Hybrid | Struct, nested-array, and null elements run natively; other element types (for example intervals), and floating-point elements when `spark.comet.exec.strictFloatingPoint=true`, route through the JVM codegen dispatcher |
| `sort_array` | ✅ | Hybrid | Struct, nested-array, floating-point, and null elements run natively; other element types (for example intervals) route through the JVM codegen dispatcher |

---

Expand Down
5 changes: 2 additions & 3 deletions docs/source/user-guide/latest/tuning/operators.md
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,5 @@ an edge case that is not of concern for many users. Setting `spark.comet.exec.st
nested cases fall back to Spark, and they can be forced back onto the native path with
`spark.comet.expression.SortOrder.allowIncompatible=true`.

`sort_array` is separate. It sorts array elements rather than ordering rows, and its elements are compared with Arrow's
raw total ordering, so `spark.comet.exec.strictFloatingPoint=true` makes it fall back even for a scalar floating-point
element type. Use `spark.comet.expression.SortArray.allowIncompatible=true` to keep it native.
`sort_array` sorts array elements rather than ordering rows. It follows Spark's floating-point ordering as well, so it
also stays native with `spark.comet.exec.strictFloatingPoint=true`.
4 changes: 4 additions & 0 deletions native/spark-expr/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -453,3 +453,7 @@ harness = false
[[bench]]
name = "nested_comparison"
harness = false

[[bench]]
name = "float_arrays"
harness = false
165 changes: 165 additions & 0 deletions native/spark-expr/benches/float_arrays.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,165 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.

//! Compare Comet's float `sort_array` and `array_remove` with DataFusion's `array_sort` and
//! `array_remove_all` on doubles without zeros or NaNs, where both return the same arrays, which
//! is checked before timing; special-value semantics belong in tests.

use arrow::array::{Array, ArrayRef, Float64Array, ListArray};
use arrow::buffer::{NullBuffer, OffsetBuffer};
use arrow::datatypes::{DataType, Field};
use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
use datafusion::common::config::ConfigOptions;
use datafusion::common::ScalarValue;
use datafusion::functions_nested::remove::array_remove_all_udf;
use datafusion::functions_nested::sort::array_sort_udf;
use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDF};
use datafusion_comet_spark_expr::{SparkArrayRemove, SparkSortArray};
use std::hint::black_box;
use std::sync::Arc;
use std::time::Duration;

const ROWS: usize = 8192;

/// The value of element `i`. Values repeat so that `array_remove` finds something to remove.
fn value(i: usize) -> f64 {
((i * 7919) % 101) as f64 * 0.5 + 1.0
}

/// `ROWS` lists of `len` doubles, with every tenth list and every seventh element null if
/// `nulls`.
fn lists(len: usize, nulls: bool) -> ArrayRef {
let values = if nulls {
Float64Array::from_iter((0..ROWS * len).map(|i| (i % 7 != 0).then(|| value(i))))
} else {
Float64Array::from_iter_values((0..ROWS * len).map(value))
};
Arc::new(ListArray::new(
Arc::new(Field::new_list_field(DataType::Float64, true)),
OffsetBuffer::from_lengths(std::iter::repeat_n(len, ROWS)),
Arc::new(values),
nulls.then(|| NullBuffer::from_iter((0..ROWS).map(|row| row % 10 != 0))),
))
}

fn args(args: Vec<ColumnarValue>, list: &ArrayRef) -> ScalarFunctionArgs {
ScalarFunctionArgs {
arg_fields: args
.iter()
.map(|arg| Arc::new(Field::new("arg", arg.data_type(), true)))
.collect(),
args,
number_rows: ROWS,
return_field: Arc::new(Field::new("result", list.data_type().clone(), true)),
config_options: Arc::new(ConfigOptions::default()),
}
}

fn criterion_benchmark(c: &mut Criterion) {
let mut group = c.benchmark_group("float_arrays");
group.sample_size(20);
group.warm_up_time(Duration::from_millis(250));
group.measurement_time(Duration::from_secs(1));
let comet_sort = ScalarUDF::new_from_impl(SparkSortArray::default());
let comet_remove = ScalarUDF::new_from_impl(SparkArrayRemove::default());
let (datafusion_sort, datafusion_remove) = (array_sort_udf(), array_remove_all_udf());
for len in [8, 50] {
for nulls in [false, true] {
let list = lists(len, nulls);
let data = format!("len={len}_null={nulls}");
let column = ColumnarValue::Array(Arc::clone(&list));
let scalar = |value: ScalarValue| ColumnarValue::Scalar(value);
let comet_sort_args = args(
vec![
column.clone(),
scalar(ScalarValue::Boolean(Some(true))),
scalar(ScalarValue::Boolean(Some(false))),
],
&list,
);
let datafusion_sort_args = args(
vec![
column.clone(),
scalar(ScalarValue::from("ASC")),
scalar(ScalarValue::from("NULLS FIRST")),
],
&list,
);
let remove_args = args(
vec![column.clone(), scalar(ScalarValue::Float64(Some(26.0)))],
&list,
);
// A value per row: the row's second element.
let row_values =
Float64Array::from_iter_values((0..ROWS).map(|row| value(row * len + 1)));
let remove_column_args = args(
vec![column, ColumnarValue::Array(Arc::new(row_values))],
&list,
);
let cases = [
(
"sort_array",
&comet_sort,
&datafusion_sort,
&comet_sort_args,
&datafusion_sort_args,
),
(
"array_remove",
&comet_remove,
&datafusion_remove,
&remove_args,
&remove_args,
),
(
"array_remove_column",
&comet_remove,
&datafusion_remove,
&remove_column_args,
&remove_column_args,
),
];
for (op, comet, datafusion, comet_args, datafusion_args) in cases {
let evaluate = |udf: &ScalarUDF, args: &ScalarFunctionArgs| {
udf.invoke_with_args(args.clone())
.unwrap()
.into_array(ROWS)
.unwrap()
};
assert_eq!(
evaluate(comet, comet_args).to_data(),
evaluate(datafusion, datafusion_args).to_data(),
"{op} {data}"
);
for (engine, udf, udf_args) in [
("comet", comet, comet_args),
("datafusion", datafusion.as_ref(), datafusion_args),
] {
group.bench_function(BenchmarkId::new(format!("{op}_{engine}"), &data), |b| {
b.iter(|| {
black_box(udf.invoke_with_args(black_box(udf_args.clone())).unwrap())
})
});
}
}
}
}
group.finish();
}

criterion_group!(benches, criterion_benchmark);
criterion_main!(benches);
Loading
Loading