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/19 09:09:55 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6989-3c7bee7067e80c8c4dfb131ca4068b4da5f88fbe created (now beae8696d1)
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-6989-3c7bee7067e80c8c4dfb131ca4068b4da5f88fbe
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at beae8696d1 fix: treat -0.0 and 0.0 as one value in arrays_overlap on Spark 4.0.5, 4.1.4 and 4.2.1 (#6989)
This branch includes the following new commits:
new beae8696d1 fix: treat -0.0 and 0.0 as one value in arrays_overlap on Spark 4.0.5, 4.1.4 and 4.2.1 (#6989)
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-comet) 01/01: fix: treat -0.0 and 0.0 as one value in arrays_overlap on Spark 4.0.5, 4.1.4 and 4.2.1 (#6989)
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-6989-3c7bee7067e80c8c4dfb131ca4068b4da5f88fbe
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit beae8696d18af00f41d0d28b3bea2114a9e612fd
Author: Andy Grove <ag...@apache.org>
AuthorDate: Mon Oct 19 09:09:34 2026 +0000
fix: treat -0.0 and 0.0 as one value in arrays_overlap on Spark 4.0.5, 4.1.4 and 4.2.1 (#6989)
* fix: treat -0.0 and 0.0 as one value in arrays_overlap on Spark 4.0.5, 4.1.4 and 4.2.1
Spark 4.0.5, 4.1.4 and 4.2.1 normalize each flat FLOAT or DOUBLE element
before arrays_overlap looks it up in its hash set of boxed values
(SPARK-59602), so -0.0 and 0.0 are one value, as they already were on
4.2.0, which normalizes the arguments in the plan (SPARK-54918). The
native kernel kept the zeros apart on every version, so on those
releases it returned false where Spark returns true.
Register a second variant, spark_arrays_overlap_normalized, whose flat
float arms key elements by normalize_float instead of canonicalize_nan,
and have CometArraysOverlap pick it wherever
ArraySetSupport.normalizesFloats holds. Earlier releases keep the
existing kernel. Both variants treat every NaN as one value and compare
nested elements in Spark's SQL ordering.
Closes #6768.
* review: name the variant in arrays_overlap errors and trim the version table
Use self.name() in the unsupported-type errors so spark_arrays_overlap_normalized reports itself,
and test it on the array and scalar paths. Reduce the native docs to the two equality policies and
point at CometArraysOverlap.nativeFunction for the version mapping. Keep one Spark version on each
side of the boundary in the routing test and the convert assertion.
---
.../expression-audits/array_funcs.md | 1 +
docs/source/contributor-guide/floating_point.md | 5 +
.../latest/compatibility/floating-point.md | 12 +
.../spark-expr/src/array_funcs/arrays_overlap.rs | 320 +++++++++++++++++----
native/spark-expr/src/comet_scalar_funcs.rs | 1 +
.../main/scala/org/apache/comet/serde/arrays.scala | 15 +-
.../apache/comet/CometArrayExpressionSuite.scala | 20 +-
7 files changed, 322 insertions(+), 52 deletions(-)
diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md
index af635270e7..87ff575307 100644
--- a/docs/source/contributor-guide/expression-audits/array_funcs.md
+++ b/docs/source/contributor-guide/expression-audits/array_funcs.md
@@ -149,6 +149,7 @@
- Spark 3.5.8 (audited 2026-05-27): baseline. `ArraysOverlap(left, right)`; three-valued logic (TRUE if any common non-null element, NULL if a null is present and no overlap is found in non-nulls, FALSE otherwise). Comet routes via `CometArraysOverlap` to the native `spark_arrays_overlap` UDF, which implements the same three-valued logic.
- 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.
+- Spark 4.0.5, 4.1.4 and 4.2.1 (release candidates, 2026-10-14): `fastEval` and its generated code normalize each flat `FLOAT` or `DOUBLE` element with `NormalizeFloatingNumbers.FLOAT_NORMALIZER` / `DOUBLE_NORMALIZER` before the `java.util.HashSet` lookup (SPARK-59602), so `-0.0` and `0.0` are one value. Spark 4.2.0 normalizes the arguments in the plan instead (SPARK-54918). Earlier releases keep the zeros apart, and every release treats all NaNs as one value. On the versions where `Arra [...]
## arrays_zip
diff --git a/docs/source/contributor-guide/floating_point.md b/docs/source/contributor-guide/floating_point.md
index 758350db8f..220a7e4ddb 100644
--- a/docs/source/contributor-guide/floating_point.md
+++ b/docs/source/contributor-guide/floating_point.md
@@ -62,6 +62,11 @@ it runs against:
(before 3.5.2, unless probing merges them). Spark 4.2.0 normalizes their arguments in the plan
(SPARK-54918). From 4.0.5, 4.1.4 and 4.2.1 they normalize while they evaluate instead
(SPARK-59602).
+- `arrays_overlap` looks up the elements of a flat array in a `java.util.HashSet` of boxed values,
+ so it follows `Double.equals` and keeps signed zeros apart up to Spark 4.0.4 and 4.1.3. The same
+ two changes make the zeros one value: Spark 4.2.0 normalizes its arguments in the plan
+ (SPARK-54918), and 4.0.5, 4.1.4 and 4.2.1 normalize each flat `FLOAT` or `DOUBLE` element before
+ the lookup (SPARK-59602). Nested elements compare in the SQL ordering on every version.
- Map construction (`ArrayBasedMapBuilder`) finds a duplicate `FLOAT` or `DOUBLE` key in a
`HashMap` of boxed values, which follows `Double.equals`, and a duplicate struct or array key in a
`TreeMap` in the SQL ordering. From Spark 4.0 it normalizes a `FLOAT` or `DOUBLE` key before the
diff --git a/docs/source/user-guide/latest/compatibility/floating-point.md b/docs/source/user-guide/latest/compatibility/floating-point.md
index 77d76f53c6..c94fc7dc68 100644
--- a/docs/source/user-guide/latest/compatibility/floating-point.md
+++ b/docs/source/user-guide/latest/compatibility/floating-point.md
@@ -148,6 +148,18 @@ differences are acceptable for your data.
The check uses the Spark version number, not the changes a build contains. A vendor build that
reports an older version but includes SPARK-54918 or SPARK-59602 still falls back.
+## `arrays_overlap`
+
+`arrays_overlap` runs natively on every Spark version and follows that version's rule for `FLOAT`
+and `DOUBLE` elements in a flat array. Spark 4.0.5+, 4.1.4+, and 4.2+ treat `-0.0` and `0.0` as one
+value (SPARK-54918, SPARK-59602), and earlier releases keep them apart. Every NaN representation is
+one value on every version, and nested arrays and structs compare their elements in Spark's order,
+in which the two zeros are equal.
+
+Comet chooses the rule by the Spark version number. On a vendor build that reports an older
+version but includes SPARK-59602, Comet returns `false` for two flat arrays whose only common value
+is a zero of opposite sign, where Spark returns `true`.
+
## `array_contains`, `array_remove` and `sort_array`
`array_contains` compares `FLOAT` and `DOUBLE` elements as Spark does: `-0.0` equals `0.0`, and all
diff --git a/native/spark-expr/src/array_funcs/arrays_overlap.rs b/native/spark-expr/src/array_funcs/arrays_overlap.rs
index 13af6636da..c761a9a51b 100644
--- a/native/spark-expr/src/array_funcs/arrays_overlap.rs
+++ b/native/spark-expr/src/array_funcs/arrays_overlap.rs
@@ -28,6 +28,12 @@
//! - true if any non-null element appears in both arrays
//! - null if no definite overlap but either array contains null elements
//! - false if no overlap and neither array contains null elements
+//!
+//! Spark looks up flat elements in a hash set of boxed values, so every NaN is one value, while
+//! `-0.0` and `0.0` are two. `spark_arrays_overlap` matches that. `spark_arrays_overlap_normalized`
+//! normalizes flat `FLOAT` and `DOUBLE` elements before the lookup, which makes the zeros one value
+//! too. `CometArraysOverlap.nativeFunction` picks the variant for the Spark version. Nested
+//! elements compare in Spark's SQL ordering in both variants, where the zeros are equal.
use arrow::array::{
make_comparator, Array, ArrayRef, AsArray, BooleanArray, GenericListArray, GenericStringArray,
@@ -46,16 +52,21 @@ use datafusion::common::{exec_err, utils::take_function_args, HashSet, Result, S
use datafusion::logical_expr::{
ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature, Volatility,
};
+use num::Float;
use std::cmp::Ordering;
use std::hash::Hash;
use std::ops::Range;
use std::sync::Arc;
-use crate::float_semantics::{canonicalize_nan, has_float_leaf, normalize_nested_floats};
+use crate::float_semantics::{
+ canonicalize_nan, has_float_leaf, normalize_float, normalize_nested_floats,
+};
#[derive(Debug, PartialEq, Eq, Hash)]
pub struct SparkArraysOverlap {
signature: Signature,
+ /// Whether `-0.0` and `0.0` are one value in a flat `FLOAT` or `DOUBLE` array.
+ normalize_floats: bool,
}
impl Default for SparkArraysOverlap {
@@ -68,13 +79,27 @@ impl SparkArraysOverlap {
pub fn new() -> Self {
Self {
signature: Signature::new(TypeSignature::Any(2), Volatility::Immutable),
+ normalize_floats: false,
+ }
+ }
+
+ /// Normalizes flat `FLOAT` and `DOUBLE` elements before matching them, so `-0.0` and `0.0` are
+ /// one value. Every NaN is one value in both variants.
+ pub fn normalized() -> Self {
+ Self {
+ normalize_floats: true,
+ ..Self::new()
}
}
}
impl ScalarUDFImpl for SparkArraysOverlap {
fn name(&self) -> &str {
- "spark_arrays_overlap"
+ if self.normalize_floats {
+ "spark_arrays_overlap_normalized"
+ } else {
+ "spark_arrays_overlap"
+ }
}
fn signature(&self) -> &Signature {
@@ -117,15 +142,15 @@ impl ScalarUDFImpl for SparkArraysOverlap {
(DataType::List(_), DataType::List(_)) => arrays_overlap_list::<i32>(
left_arr.as_any().downcast_ref().unwrap(),
right_arr.as_any().downcast_ref().unwrap(),
+ self.normalize_floats,
)?,
(DataType::LargeList(_), DataType::LargeList(_)) => arrays_overlap_list::<i64>(
left_arr.as_any().downcast_ref().unwrap(),
right_arr.as_any().downcast_ref().unwrap(),
+ self.normalize_floats,
)?,
(l, r) => {
- return exec_err!(
- "spark_arrays_overlap does not support types '{l}' and '{r}'"
- )
+ return exec_err!("{} does not support types '{l}' and '{r}'", self.name())
}
};
Ok(ColumnarValue::Array(result))
@@ -138,15 +163,15 @@ impl ScalarUDFImpl for SparkArraysOverlap {
(DataType::List(_), DataType::List(_)) => arrays_overlap_list::<i32>(
left_arr.as_any().downcast_ref().unwrap(),
right_arr.as_any().downcast_ref().unwrap(),
+ self.normalize_floats,
)?,
(DataType::LargeList(_), DataType::LargeList(_)) => arrays_overlap_list::<i64>(
left_arr.as_any().downcast_ref().unwrap(),
right_arr.as_any().downcast_ref().unwrap(),
+ self.normalize_floats,
)?,
(l, r) => {
- return exec_err!(
- "spark_arrays_overlap does not support types '{l}' and '{r}'"
- )
+ return exec_err!("{} does not support types '{l}' and '{r}'", self.name())
}
};
let scalar = ScalarValue::try_from_array(&result, 0)?;
@@ -163,9 +188,12 @@ impl ScalarUDFImpl for SparkArraysOverlap {
/// - true if any non-null element appears in both arrays
/// - null if no definite overlap but either array contains null elements
/// - false otherwise
+///
+/// With `normalize_floats`, `-0.0` and `0.0` are one value in a flat `FLOAT` or `DOUBLE` array.
fn arrays_overlap_list<OffsetSize: OffsetSizeTrait>(
left: &GenericListArray<OffsetSize>,
right: &GenericListArray<OffsetSize>,
+ normalize_floats: bool,
) -> Result<ArrayRef> {
let left_values = left.values();
let right_values = right.values();
@@ -189,6 +217,18 @@ fn arrays_overlap_list<OffsetSize: OffsetSizeTrait>(
)
};
}
+ macro_rules! float_fast_path {
+ ($t:ty) => {
+ if normalize_floats {
+ flat_fast_path!(
+ NormalizedFloats(left_values.as_primitive::<$t>()),
+ NormalizedFloats(right_values.as_primitive::<$t>())
+ )
+ } else {
+ primitive_fast_path!($t)
+ }
+ };
+ }
match left_values.data_type() {
DataType::Boolean => flat_fast_path!(left_values.as_boolean(), right_values.as_boolean()),
@@ -200,8 +240,8 @@ fn arrays_overlap_list<OffsetSize: OffsetSizeTrait>(
DataType::UInt16 => primitive_fast_path!(UInt16Type),
DataType::UInt32 => primitive_fast_path!(UInt32Type),
DataType::UInt64 => primitive_fast_path!(UInt64Type),
- DataType::Float32 => primitive_fast_path!(Float32Type),
- DataType::Float64 => primitive_fast_path!(Float64Type),
+ DataType::Float32 => float_fast_path!(Float32Type),
+ DataType::Float64 => float_fast_path!(Float64Type),
DataType::Date32 => primitive_fast_path!(Date32Type),
DataType::Date64 => primitive_fast_path!(Date64Type),
DataType::Decimal128(_, _) => primitive_fast_path!(Decimal128Type),
@@ -348,6 +388,30 @@ impl<'a, S: OffsetSizeTrait> KeyedValues<'a> for &'a GenericStringArray<S> {
}
}
+/// Flat `FLOAT` or `DOUBLE` elements keyed after [`normalize_float`], so `-0.0` and `0.0` share a
+/// key as well as every NaN.
+struct NormalizedFloats<'a, T: ArrowPrimitiveType>(&'a PrimitiveArray<T>);
+
+// Derived impls would require `T: Copy`, which Arrow's type markers don't implement.
+impl<T: ArrowPrimitiveType> Clone for NormalizedFloats<'_, T> {
+ fn clone(&self) -> Self {
+ *self
+ }
+}
+
+impl<T: ArrowPrimitiveType> Copy for NormalizedFloats<'_, T> {}
+
+impl<'a, T: ArrowPrimitiveType> KeyedValues<'a> for NormalizedFloats<'a, T>
+where
+ T::Native: OverlapKey + Float,
+{
+ type Key = <T::Native as OverlapKey>::Key;
+
+ fn key_at(&self, i: usize) -> Option<Self::Key> {
+ (!self.0.is_null(i)).then(|| normalize_float(self.0.value(i)).overlap_key())
+ }
+}
+
/// Above this many pairwise comparisons a hash probe beats the nested scan.
const NESTED_SCAN_BUDGET: usize = 256;
@@ -407,9 +471,8 @@ fn normalize_list_element_floats<OffsetSize: OffsetSizeTrait>(
/// Fallback for nested and otherwise unhandled element types.
///
-/// note: Spark's flat arrays_overlap (HashSet<Double>) treats -0.0 and 0.0 as different,
-/// only the nested path here treats them as equal. this normalization can't move into the
-/// flat fast path in arrays_overlap_list without breaking that difference.
+/// Spark compares nested elements in its SQL ordering on every version, so this path normalizes
+/// their floats whether or not the flat fast path in arrays_overlap_list does.
fn arrays_overlap_list_generic<OffsetSize: OffsetSizeTrait>(
left: &GenericListArray<OffsetSize>,
right: &GenericListArray<OffsetSize>,
@@ -538,11 +601,12 @@ fn needs_comparator(dt: &DataType) -> bool {
mod tests {
use super::*;
use arrow::array::{
- Float64Array, Float64Builder, Int32Array, Int32Builder, ListArray, ListBuilder,
- StructArray, StructBuilder,
+ Float64Array, Float64Builder, Int32Array, Int32Builder, LargeListArray, ListArray,
+ ListBuilder, StructArray, StructBuilder,
};
use arrow::buffer::{NullBuffer, OffsetBuffer};
use arrow::datatypes::{Field, Fields};
+ use datafusion::config::ConfigOptions;
fn make_list_array(
values: &Int32Array,
@@ -563,7 +627,7 @@ mod tests {
let left = make_list_array(&Int32Array::from(vec![1, 2, 3]), &[0, 3], None);
let right = make_list_array(&Int32Array::from(vec![3, 4, 5]), &[0, 3], None);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.value(0));
assert!(result.is_valid(0));
@@ -576,7 +640,7 @@ mod tests {
let left = make_list_array(&Int32Array::from(vec![1, 2]), &[0, 2], None);
let right = make_list_array(&Int32Array::from(vec![3, 4]), &[0, 2], None);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(!result.value(0));
assert!(result.is_valid(0));
@@ -605,6 +669,10 @@ mod tests {
.map(|value| Some(value as f32))
.chain([Some(-0.0)])
.collect::<Vec<_>>();
+ let hash_sign_right = (17..=32)
+ .map(|value| Some(value as f32))
+ .chain([Some(-1.0)])
+ .collect::<Vec<_>>();
let left = ListArray::from_iter_primitive::<Float32Type, _, _>([
Some(vec![Some(positive_nan)]),
@@ -615,6 +683,8 @@ mod tests {
Some(vec![Some(positive_nan), None]),
Some(vec![Some(0.0), None]),
Some(hash_nan_left),
+ Some(hash_zero_left.clone()),
+ Some(vec![Some(-1.0)]),
Some(hash_zero_left),
]);
let right = ListArray::from_iter_primitive::<Float32Type, _, _>([
@@ -627,11 +697,13 @@ mod tests {
Some(vec![Some(-0.0)]),
Some(hash_nan_right),
Some(hash_zero_right),
+ Some(vec![Some(1.0)]),
+ Some(hash_sign_right),
]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
- let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
- let expected = BooleanArray::from(vec![
+ // Every NaN is one value in both variants, and only the normalized one also makes the
+ // zeros one value. The last two rows check that other values keep their sign.
+ let boxed = BooleanArray::from(vec![
Some(true),
Some(true),
Some(true),
@@ -641,8 +713,30 @@ mod tests {
None,
Some(true),
Some(false),
+ Some(false),
+ Some(false),
+ ]);
+ let normalized = BooleanArray::from(vec![
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(false),
+ Some(false),
]);
- assert_eq!(result, &expected);
+ for (normalize_floats, expected) in [(false, boxed), (true, normalized)] {
+ let result = arrays_overlap_list::<i32>(&left, &right, normalize_floats)?;
+ assert_eq!(
+ result.as_boolean(),
+ &expected,
+ "normalize_floats: {normalize_floats}"
+ );
+ }
Ok(())
}
@@ -668,6 +762,10 @@ mod tests {
.map(|value| Some(value as f64))
.chain([Some(-0.0)])
.collect::<Vec<_>>();
+ let hash_sign_right = (17..=32)
+ .map(|value| Some(value as f64))
+ .chain([Some(-1.0)])
+ .collect::<Vec<_>>();
let left = ListArray::from_iter_primitive::<Float64Type, _, _>([
Some(vec![Some(positive_nan)]),
@@ -678,6 +776,8 @@ mod tests {
Some(vec![Some(positive_nan), None]),
Some(vec![Some(0.0), None]),
Some(hash_nan_left),
+ Some(hash_zero_left.clone()),
+ Some(vec![Some(-1.0)]),
Some(hash_zero_left),
]);
let right = ListArray::from_iter_primitive::<Float64Type, _, _>([
@@ -690,11 +790,13 @@ mod tests {
Some(vec![Some(-0.0)]),
Some(hash_nan_right),
Some(hash_zero_right),
+ Some(vec![Some(1.0)]),
+ Some(hash_sign_right),
]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
- let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
- let expected = BooleanArray::from(vec![
+ // Every NaN is one value in both variants, and only the normalized one also makes the
+ // zeros one value. The last two rows check that other values keep their sign.
+ let boxed = BooleanArray::from(vec![
Some(true),
Some(true),
Some(true),
@@ -704,18 +806,140 @@ mod tests {
None,
Some(true),
Some(false),
+ Some(false),
+ Some(false),
]);
- assert_eq!(result, &expected);
+ let normalized = BooleanArray::from(vec![
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(true),
+ Some(false),
+ Some(false),
+ ]);
+ for (normalize_floats, expected) in [(false, boxed), (true, normalized)] {
+ let result = arrays_overlap_list::<i32>(&left, &right, normalize_floats)?;
+ assert_eq!(
+ result.as_boolean(),
+ &expected,
+ "normalize_floats: {normalize_floats}"
+ );
+ }
Ok(())
}
+ /// Runs `udf` over two arguments, giving one value per row.
+ fn invoke(udf: &SparkArraysOverlap, args: [ColumnarValue; 2], rows: usize) -> Result<ArrayRef> {
+ let arg_fields = args
+ .iter()
+ .map(|arg| Arc::new(Field::new("arg", arg.data_type(), true)))
+ .collect();
+ udf.invoke_with_args(ScalarFunctionArgs {
+ args: args.into(),
+ arg_fields,
+ number_rows: rows,
+ return_field: Arc::new(Field::new("result", DataType::Boolean, true)),
+ config_options: Arc::new(ConfigOptions::default()),
+ })?
+ .into_array(rows)
+ }
+
+ #[test]
+ fn test_normalized_variant_makes_flat_signed_zeros_one_value() -> Result<()> {
+ // Row 0 holds the two zeros, and row 1 two values that differ only in sign.
+ let left = [Some(vec![Some(0.0)]), Some(vec![Some(1.0)])];
+ let right = [Some(vec![Some(-0.0)]), Some(vec![Some(-1.0)])];
+ let lists: [(ArrayRef, ArrayRef); 2] = [
+ (
+ Arc::new(ListArray::from_iter_primitive::<Float64Type, _, _>(
+ left.clone(),
+ )),
+ Arc::new(ListArray::from_iter_primitive::<Float64Type, _, _>(
+ right.clone(),
+ )),
+ ),
+ (
+ Arc::new(LargeListArray::from_iter_primitive::<Float64Type, _, _>(
+ left,
+ )),
+ Arc::new(LargeListArray::from_iter_primitive::<Float64Type, _, _>(
+ right,
+ )),
+ ),
+ ];
+
+ for (udf, name, zeros_match) in [
+ (SparkArraysOverlap::new(), "spark_arrays_overlap", false),
+ (
+ SparkArraysOverlap::normalized(),
+ "spark_arrays_overlap_normalized",
+ true,
+ ),
+ ] {
+ assert_eq!(udf.name(), name);
+ for (left, right) in &lists {
+ let columns = [
+ ColumnarValue::Array(Arc::clone(left)),
+ ColumnarValue::Array(Arc::clone(right)),
+ ];
+ let result = invoke(&udf, columns, 2)?;
+ assert_eq!(
+ result.as_boolean(),
+ &BooleanArray::from(vec![zeros_match, false]),
+ "{name} over {} columns",
+ left.data_type()
+ );
+
+ let scalars = [
+ ColumnarValue::Scalar(ScalarValue::try_from_array(left, 0)?),
+ ColumnarValue::Scalar(ScalarValue::try_from_array(right, 0)?),
+ ];
+ let result = invoke(&udf, scalars, 1)?;
+ assert_eq!(
+ result.as_boolean(),
+ &BooleanArray::from(vec![zeros_match]),
+ "{name} over {} scalars",
+ left.data_type()
+ );
+ }
+ }
+ Ok(())
+ }
+
+ #[test]
+ fn test_unsupported_types_error_names_the_variant() {
+ for udf in [SparkArraysOverlap::new(), SparkArraysOverlap::normalized()] {
+ let column = Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef;
+ let arrays = [
+ ColumnarValue::Array(Arc::clone(&column)),
+ ColumnarValue::Array(column),
+ ];
+ let scalars = [
+ ColumnarValue::Scalar(ScalarValue::Int32(Some(1))),
+ ColumnarValue::Scalar(ScalarValue::Int32(Some(2))),
+ ];
+ for (args, rows) in [(arrays, 2), (scalars, 1)] {
+ let message = invoke(&udf, args, rows).unwrap_err().to_string();
+ assert!(
+ message.contains(&format!("{} does not support types", udf.name())),
+ "{message}"
+ );
+ }
+ }
+ }
+
#[test]
fn test_null_only_overlap() -> Result<()> {
// [1, NULL] vs [NULL, 2] => null (no definite overlap, but nulls present)
let left = make_list_array(&Int32Array::from(vec![Some(1), None]), &[0, 2], None);
let right = make_list_array(&Int32Array::from(vec![None, Some(2)]), &[0, 2], None);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_null(0));
Ok(())
@@ -727,7 +951,7 @@ mod tests {
let left = make_list_array(&Int32Array::from(vec![Some(1), None]), &[0, 2], None);
let right = make_list_array(&Int32Array::from(vec![1, 2]), &[0, 2], None);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.value(0));
assert!(result.is_valid(0));
@@ -744,7 +968,7 @@ mod tests {
);
let right = make_list_array(&Int32Array::from(Vec::<i32>::new()), &[0, 0], None);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(!result.value(0));
assert!(result.is_valid(0));
@@ -761,7 +985,7 @@ mod tests {
);
let right = make_list_array(&Int32Array::from(vec![1, 2]), &[0, 2], None);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_null(0));
Ok(())
@@ -773,7 +997,7 @@ mod tests {
let left = make_list_array(&Int32Array::from(vec![None::<i32>]), &[0, 1], None);
let right = make_list_array(&Int32Array::from(vec![None::<i32>]), &[0, 1], None);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_null(0));
Ok(())
@@ -799,7 +1023,7 @@ mod tests {
None,
);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(
result.is_null(0),
@@ -815,7 +1039,7 @@ mod tests {
let left = make_list_array(&Int32Array::from(vec![Some(3), None]), &[0, 2], None);
let right = make_list_array(&Int32Array::from(vec![1, 2]), &[0, 2], None);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_null(0));
Ok(())
@@ -864,13 +1088,13 @@ mod tests {
// NaN matches itself, and signed zeros are equal, matching Spark.
let left = make_nested_float_list(&[&[f64::NAN]]);
let right = make_nested_float_list(&[&[f64::NAN]]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.value(0));
let left = make_nested_float_list(&[&[0.0]]);
let right = make_nested_float_list(&[&[-0.0]]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.value(0));
Ok(())
@@ -881,7 +1105,7 @@ mod tests {
// [[-NaN]] vs [[NaN]] => true
let left = make_nested_float_list(&[&[-f64::NAN]]);
let right = make_nested_float_list(&[&[f64::NAN]]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.value(0));
Ok(())
@@ -899,7 +1123,7 @@ mod tests {
Some(vec![Some(5), Some(6)]),
]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(result.value(0));
@@ -912,7 +1136,7 @@ mod tests {
let left = make_nested_list(vec![Some(vec![Some(1), Some(2)])]);
let right = make_nested_list(vec![Some(vec![Some(3), Some(4)])]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(!result.value(0));
@@ -925,7 +1149,7 @@ mod tests {
let left = make_nested_list(vec![Some(vec![Some(1), None])]);
let right = make_nested_list(vec![Some(vec![Some(1), None])]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(result.value(0));
@@ -941,7 +1165,7 @@ mod tests {
Some(vec![Some(3), Some(4)]),
]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(!result.value(0));
@@ -954,7 +1178,7 @@ mod tests {
let left = make_nested_list(vec![Some(vec![None])]);
let right = make_nested_list(vec![Some(vec![None])]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(
result.is_valid(0),
@@ -976,7 +1200,7 @@ mod tests {
]);
let right = make_nested_list(vec![Some(vec![Some(1), Some(2)])]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(result.value(0));
@@ -1024,7 +1248,7 @@ mod tests {
let left = make_struct_list(vec![Some((Some(1), Some(2))), Some((Some(3), Some(4)))]);
let right = make_struct_list(vec![Some((Some(3), Some(4))), Some((Some(5), Some(6)))]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(result.value(0));
@@ -1037,7 +1261,7 @@ mod tests {
let left = make_struct_list(vec![Some((Some(1), Some(2)))]);
let right = make_struct_list(vec![Some((Some(3), Some(4)))]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(!result.value(0));
@@ -1050,7 +1274,7 @@ mod tests {
let left = make_struct_list(vec![Some((Some(1), None))]);
let right = make_struct_list(vec![Some((Some(1), None))]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(result.value(0));
@@ -1063,7 +1287,7 @@ mod tests {
let left = make_struct_list(vec![Some((Some(1), Some(2))), Some((Some(1), None))]);
let right = make_struct_list(vec![Some((Some(1), Some(2)))]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(result.value(0));
@@ -1117,7 +1341,7 @@ mod tests {
let left = make_list(vec![1.0, 2.0]);
let right = make_list(vec![1.0, 2.0]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.value(0));
assert!(result.value(1));
@@ -1130,7 +1354,7 @@ mod tests {
let left = make_struct_float_list(vec![Some(-0.0)]);
let right = make_struct_float_list(vec![Some(0.0)]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_valid(0));
assert!(result.value(0));
@@ -1143,7 +1367,7 @@ mod tests {
let left = make_struct_list(vec![None]);
let right = make_struct_list(vec![Some((Some(1), Some(2)))]);
- let result = arrays_overlap_list::<i32>(&left, &right)?;
+ let result = arrays_overlap_list::<i32>(&left, &right, false)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert!(result.is_null(0));
Ok(())
diff --git a/native/spark-expr/src/comet_scalar_funcs.rs b/native/spark-expr/src/comet_scalar_funcs.rs
index 504db52f6d..ce686557ae 100644
--- a/native/spark-expr/src/comet_scalar_funcs.rs
+++ b/native/spark-expr/src/comet_scalar_funcs.rs
@@ -349,6 +349,7 @@ fn all_scalar_functions() -> Vec<Arc<ScalarUDF>> {
Arc::new(ScalarUDF::new_from_impl(SparkFloatArrayContains::default())),
Arc::new(ScalarUDF::new_from_impl(SparkArraySlice::default())),
Arc::new(ScalarUDF::new_from_impl(SparkArraysOverlap::default())),
+ Arc::new(ScalarUDF::new_from_impl(SparkArraysOverlap::normalized())),
Arc::new(ScalarUDF::new_from_impl(SparkSortArray::default())),
Arc::new(ScalarUDF::new_from_impl(SparkContains::default())),
Arc::new(ScalarUDF::new_from_impl(SparkDateDiff::default())),
diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
index e8778b335e..3b0edd3172 100644
--- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
@@ -363,6 +363,16 @@ object CometArraysOverlap
override def getSupportLevel(expr: ArraysOverlap): SupportLevel =
ArrayElementEqualitySupport.getSupportLevel("arrays_overlap", expr.left.dataType)
+ /**
+ * The native variant for a Spark version. Where ArraySetSupport.normalizesFloats holds, Spark
+ * normalizes flat FLOAT and DOUBLE elements, so -0.0 and 0.0 are one value, and the
+ * `_normalized` variant matches them that way. Earlier releases keep the zeros apart. Both
+ * variants treat every NaN as one value and compare nested elements in Spark's SQL ordering.
+ */
+ def nativeFunction(version: String): String =
+ if (ArraySetSupport.normalizesFloats(version)) "spark_arrays_overlap_normalized"
+ else "spark_arrays_overlap"
+
override def convert(
expr: ArraysOverlap,
inputs: Seq[Attribute],
@@ -371,7 +381,7 @@ object CometArraysOverlap
val rightArrayExprProto = exprToProtoInternal(expr.right, inputs, binding)
val arraysOverlapScalarExpr = scalarFunctionExprToProtoWithReturnType(
- "spark_arrays_overlap",
+ nativeFunction(SPARK_VERSION),
BooleanType,
false,
leftArrayExprProto,
@@ -626,7 +636,8 @@ private[comet] object ArraySetSupport {
// Spark 4.2.0 normalizes the arguments of these functions in the plan (SPARK-54918), and 4.0.5,
// 4.1.4 and 4.2.1 normalize while evaluating them (SPARK-59602). Either way, Spark treats -0.0
// and 0.0, and every NaN, as one value at any depth, which the spark_ variants match. Earlier
- // releases keep -0.0 and 0.0 apart in a flat array. The check reads the version rather than a
+ // releases keep -0.0 and 0.0 apart in a flat array. Both changes cover arrays_overlap too (see
+ // CometArraysOverlap.nativeFunction). The check reads the version rather than a
// KnownFloatingPointNormalized marker, because SPARK-59602 adds no marker, and SPARK-54918
// normalizes CreateArray, If, CaseWhen and Coalesce without wrapping the resulting array.
def normalizesFloats(version: String): Boolean =
diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
index fd7730aa9b..6ca4b80538 100644
--- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
@@ -24,7 +24,7 @@ import scala.util.Random
import org.apache.hadoop.fs.Path
import org.apache.spark.SparkException
import org.apache.spark.sql.CometTestBase
-import org.apache.spark.sql.catalyst.expressions.{ArrayAppend, ArrayDistinct, ArrayExcept, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayRepeat, ArrayUnion}
+import org.apache.spark.sql.catalyst.expressions.{ArrayAppend, ArrayDistinct, ArrayExcept, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayRepeat, ArraysOverlap, ArrayUnion}
import org.apache.spark.sql.catalyst.expressions.{ArrayContains, ArrayRemove}
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, CreateArray, ElementAt, Literal, MonotonicallyIncreasingID}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
@@ -34,7 +34,7 @@ import org.apache.spark.sql.types.{ArrayType, DataType, DoubleType, FloatType, I
import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus}
import org.apache.comet.DataTypeSupport.isComplexType
-import org.apache.comet.serde.{ArrayElementEqualitySupport, ArraySetSupport, CometArrayDistinct, CometArrayExcept, CometArrayJoin, CometArrayRemove, CometArrayReverse, CometArrayUnion, CometFlatten, Compatible, ExprOuterClass, Incompatible}
+import org.apache.comet.serde.{ArrayElementEqualitySupport, ArraySetSupport, CometArrayDistinct, CometArrayExcept, CometArrayJoin, CometArrayRemove, CometArrayReverse, CometArraysOverlap, CometArrayUnion, CometFlatten, Compatible, ExprOuterClass, Incompatible}
import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions}
class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
@@ -771,6 +771,22 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp
assert(nullableDelimiter.isDefined && !nullableDelimiter.get.hasIf)
}
+ test("arrays_overlap native variant by Spark version") {
+ // "array set signed-zero Spark patch versions" pins the version boundaries. Where Spark
+ // normalizes, -0.0 and 0.0 are one value in a flat FLOAT or DOUBLE array, which no pinned
+ // Spark profile can show: 4.2.0 normalizes the arguments in the plan, before the native
+ // kernel sees them.
+ assert(CometArraysOverlap.nativeFunction("4.1.3") == "spark_arrays_overlap")
+ assert(CometArraysOverlap.nativeFunction("4.1.4") == "spark_arrays_overlap_normalized")
+ // convert asks for the variant of the Spark version it runs on.
+ val array = AttributeReference("a", ArrayType(DoubleType))()
+ val proto =
+ CometArraysOverlap.convert(ArraysOverlap(array, array), Seq(array), binding = false)
+ assert(
+ proto.map(_.getScalarFunc.getFunc) ==
+ Some(CometArraysOverlap.nativeFunction(org.apache.spark.SPARK_VERSION)))
+ }
+
test("arrays_overlap") {
Seq(true, false).foreach { dictionaryEnabled =>
withTempDir { dir =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org