From a43bface742200b4f20cde7fbb02af4e4c919bab Mon Sep 17 00:00:00 2001 From: peterxcli Date: Sat, 1 Aug 2026 23:16:33 +0800 Subject: [PATCH 1/5] perf: reuse nested array comparator across rows --- native/spark-expr/benches/arrays_overlap.rs | 57 ++++++++- .../src/array_funcs/array_position.rs | 5 +- .../src/array_funcs/arrays_overlap.rs | 110 ++++++++++++------ 3 files changed, 133 insertions(+), 39 deletions(-) diff --git a/native/spark-expr/benches/arrays_overlap.rs b/native/spark-expr/benches/arrays_overlap.rs index c20efdd8b67..c6aa11fba42 100644 --- a/native/spark-expr/benches/arrays_overlap.rs +++ b/native/spark-expr/benches/arrays_overlap.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use arrow::array::{ArrayRef, Int32Array, ListArray, StringArray}; +use arrow::array::{ArrayRef, Int32Array, ListArray, StringArray, StructArray}; use arrow::buffer::{NullBuffer, OffsetBuffer}; use arrow::datatypes::{DataType, Field}; use criterion::{criterion_group, criterion_main, Criterion}; @@ -71,6 +71,41 @@ fn string_lists(rows: usize, elems_per_row: usize, offset: usize) -> (ArrayRef, ) } +fn nested_int_lists(rows: usize, elems_per_row: usize) -> (ArrayRef, ArrayRef) { + let total = rows * elems_per_row; + let build = |offset: i32| { + let values: ArrayRef = Arc::new(Int32Array::from_iter_values( + (0..total).flat_map(|i| [0, 1, 2, i as i32 + offset]), + )); + list_of(values, total, 4) + }; + ( + list_of(build(0), rows, elems_per_row), + list_of(build(total as i32), rows, elems_per_row), + ) +} + +fn struct_lists(rows: usize, elems_per_row: usize) -> (ArrayRef, ArrayRef) { + let total = rows * elems_per_row; + let build = |offset: i32| -> ArrayRef { + let first: ArrayRef = Arc::new(Int32Array::from_value(0, total)); + let second: ArrayRef = Arc::new(Int32Array::from_iter_values( + (0..total).map(|i| i as i32 + offset), + )); + Arc::new(StructArray::from(vec![ + (Arc::new(Field::new("first", DataType::Int32, false)), first), + ( + Arc::new(Field::new("second", DataType::Int32, false)), + second, + ), + ])) + }; + ( + list_of(build(0), rows, elems_per_row), + list_of(build(total as i32), rows, elems_per_row), + ) +} + fn invoke(udf: &SparkArraysOverlap, left: &ArrayRef, right: &ArrayRef) -> ColumnarValue { udf.invoke_with_args(ScalarFunctionArgs { args: vec![ @@ -113,6 +148,26 @@ fn criterion_benchmark(c: &mut Criterion) { c.bench_function("spark_arrays_overlap: utf8 long lists", |b| { b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) }); + + let (left, right) = nested_int_lists(rows, 8); + c.bench_function("spark_arrays_overlap: nested int32 short lists", |b| { + b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) + }); + + let (left, right) = nested_int_lists(64, 64); + c.bench_function("spark_arrays_overlap: nested int32 long lists", |b| { + b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) + }); + + let (left, right) = struct_lists(rows, 8); + c.bench_function("spark_arrays_overlap: nested struct short lists", |b| { + b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) + }); + + let (left, right) = struct_lists(64, 64); + c.bench_function("spark_arrays_overlap: nested struct long lists", |b| { + b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) + }); } criterion_group!(benches, criterion_benchmark); diff --git a/native/spark-expr/src/array_funcs/array_position.rs b/native/spark-expr/src/array_funcs/array_position.rs index 191091aabf9..25f0fea0c4d 100644 --- a/native/spark-expr/src/array_funcs/array_position.rs +++ b/native/spark-expr/src/array_funcs/array_position.rs @@ -301,8 +301,9 @@ mod tests { #[test] fn test_nested_float_and_null_position() -> DataFusionResult<()> { - // Arrow and the previous ScalarValue fallback distinguish signed zeros, so the second - // row matches at position 2 rather than position 1. + // Signed-zero equality does not yet match Spark, so the second row matches at position 2; + // see https://github.com/apache/datafusion-comet/issues/5191. + // NaN and inner-null equality match Spark. let values = ListArray::from_iter_primitive::([ Some(vec![Some(1.0)]), Some(vec![Some(f64::NAN)]), diff --git a/native/spark-expr/src/array_funcs/arrays_overlap.rs b/native/spark-expr/src/array_funcs/arrays_overlap.rs index bd75a6ddccb..5500a71254f 100644 --- a/native/spark-expr/src/array_funcs/arrays_overlap.rs +++ b/native/spark-expr/src/array_funcs/arrays_overlap.rs @@ -388,11 +388,60 @@ where } } +/// Row overlap for nested element types using one comparator for the full child arrays. +fn nested_row_overlap<'a>( + left: &'a ArrayRef, + right: &'a ArrayRef, + comparator: &'a dyn Fn(usize, usize) -> Ordering, +) -> impl FnMut(Range, Range) -> bool + 'a { + move |left_range, right_range| { + let (probe, probe_range, search, search_range, probe_is_left) = + if left_range.len() <= right_range.len() { + (left, left_range, right, right_range, true) + } else { + (right, right_range, left, left_range, false) + }; + + for pi in probe_range { + if probe.is_null(pi) { + continue; + } + for si in search_range.clone() { + if search.is_null(si) { + continue; + } + let (li, ri) = if probe_is_left { (pi, si) } else { (si, pi) }; + if comparator(li, ri) == Ordering::Equal { + return true; + } + } + } + false + } +} + /// Fallback for nested and otherwise unhandled element types. fn arrays_overlap_list_generic( left: &GenericListArray, right: &GenericListArray, ) -> Result { + let left_values = left.values(); + let right_values = right.values(); + if left_values.data_type() == right_values.data_type() + && needs_comparator(left_values.data_type()) + { + let comparator = make_comparator( + left_values.as_ref(), + right_values.as_ref(), + SortOptions::default(), + )?; + return Ok(overlap_rows( + left, + right, + nested_row_overlap(left_values, right_values, comparator.as_ref()), + )); + } + let len = left.len(); let mut builder = BooleanArray::builder(len); @@ -428,26 +477,12 @@ fn arrays_overlap_list_generic( (&right_values, &left_values) }; - let comparator = if needs_comparator(probe.data_type()) { - Some(make_comparator( - probe.as_ref(), - search.as_ref(), - SortOptions::default(), - )?) - } else { - None - }; - for pi in 0..probe.len() { if probe.is_null(pi) { has_null = true; continue; } - let (found, null_eq) = if let Some(comparator) = &comparator { - find_in_array_nested(pi, search, comparator.as_ref()) - } else { - find_in_array_flat(probe, pi, search)? - }; + let (found, null_eq) = find_in_array_flat(probe, pi, search)?; if null_eq { has_null = true; } @@ -477,25 +512,6 @@ fn find_in_array_flat(probe: &ArrayRef, pi: usize, search: &ArrayRef) -> Result< Ok((eq_result.true_count() > 0, eq_result.null_count() > 0)) } -/// Element-by-element search using Arrow's nested comparator. -fn find_in_array_nested( - pi: usize, - search: &ArrayRef, - comparator: &dyn Fn(usize, usize) -> Ordering, -) -> (bool, bool) { - let mut has_null = false; - for si in 0..search.len() { - if search.is_null(si) { - has_null = true; - continue; - } - if comparator(pi, si) == Ordering::Equal { - return (true, has_null); - } - } - (false, has_null) -} - fn needs_comparator(dt: &DataType) -> bool { matches!( dt, @@ -706,14 +722,14 @@ mod tests { #[test] fn test_nested_float_total_order() -> Result<()> { - // Preserve the existing Arrow total-order behavior: NaN matches itself, while signed - // zeros are distinct. + // NaN equality matches Spark. let left = make_nested_float_list(&[&[f64::NAN]]); let right = make_nested_float_list(&[&[f64::NAN]]); let result = arrays_overlap_list::(&left, &right)?; let result = result.as_any().downcast_ref::().unwrap(); assert!(result.value(0)); + // Signed-zero equality does not yet match Spark; see #5191. let left = make_nested_float_list(&[&[0.0]]); let right = make_nested_float_list(&[&[-0.0]]); let result = arrays_overlap_list::(&left, &right)?; @@ -722,6 +738,28 @@ mod tests { Ok(()) } + #[test] + fn test_nested_array_sliced_offsets_and_probe_swap() -> Result<()> { + let make_rows = |rows: &[&[&[i32]]]| { + let mut builder = ListBuilder::new(ListBuilder::new(Int32Builder::new())); + for row in rows { + for element in *row { + builder.values().values().append_slice(element); + builder.values().append(true); + } + builder.append(true); + } + builder.finish() + }; + let left = make_rows(&[&[&[999]], &[&[10]], &[&[50], &[60], &[70]]]).slice(1, 2); + let right = make_rows(&[&[&[999]], &[&[20], &[30], &[40]], &[&[60]]]).slice(1, 2); + + let result = arrays_overlap_list::(&left, &right)?; + let result = result.as_any().downcast_ref::().unwrap(); + assert_eq!(result, &BooleanArray::from(vec![false, true])); + Ok(()) + } + #[test] fn test_nested_array_basic_overlap() -> Result<()> { // [[1,2], [3,4]] vs [[3,4], [5,6]] => true From 5c730490cf75b78dfc8c5ba7f51095886fef6b9c Mon Sep 17 00:00:00 2001 From: peterxcli Date: Thu, 6 Aug 2026 21:44:06 +0800 Subject: [PATCH 2/5] fix: address nested comparator review --- docs/source/user-guide/latest/expressions.md | 4 +- native/spark-expr/benches/arrays_overlap.rs | 17 ++-- .../src/array_funcs/arrays_overlap.rs | 80 ++++++++++--------- .../scala/org/apache/comet/serde/arrays.scala | 33 +++++++- .../expressions/array/arrays_overlap.sql | 10 +++ 5 files changed, 97 insertions(+), 47 deletions(-) diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index fd4324babc6..e857a46398a 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -158,12 +158,12 @@ The tables below list every Spark built-in expression with its current status. | `array_join` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the incompatible native path is opt-in via allowIncompatible ([details](compatibility/expressions/array.md)) | | `array_max` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | | `array_min` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | -| `array_position` | ✅ | Native | Binary/struct/map/null elements fall back | +| `array_position` | ✅ | Native | Binary/struct/map/null elements fall back; nested floating-point signed-zero handling differs ([#5191](https://github.com/apache/datafusion-comet/issues/5191)) | | `array_prepend` | ✅ | — | | | `array_remove` | ✅ | Native | | | `array_repeat` | ✅ | Native | | | `array_union` | ✅ | Native | NaN/signed-zero handling may differ ([details](compatibility/floating-point.md)) | -| `arrays_overlap` | ✅ | Native | | +| `arrays_overlap` | ✅ | Native | Nested floating-point signed-zero handling differs ([#5191](https://github.com/apache/datafusion-comet/issues/5191)) | | `arrays_zip` | ✅ | Native | | | `element_at` | ✅ | Native | | | `flatten` | ✅ | Native | Binary/struct/map elements fall back | diff --git a/native/spark-expr/benches/arrays_overlap.rs b/native/spark-expr/benches/arrays_overlap.rs index c6aa11fba42..67f5b56d138 100644 --- a/native/spark-expr/benches/arrays_overlap.rs +++ b/native/spark-expr/benches/arrays_overlap.rs @@ -71,17 +71,17 @@ fn string_lists(rows: usize, elems_per_row: usize, offset: usize) -> (ArrayRef, ) } -fn nested_int_lists(rows: usize, elems_per_row: usize) -> (ArrayRef, ArrayRef) { +fn nested_int_lists(rows: usize, elems_per_row: usize, offset: i32) -> (ArrayRef, ArrayRef) { let total = rows * elems_per_row; - let build = |offset: i32| { + let build = |value_offset: i32| { let values: ArrayRef = Arc::new(Int32Array::from_iter_values( - (0..total).flat_map(|i| [0, 1, 2, i as i32 + offset]), + (0..total).flat_map(|i| [0, 1, 2, i as i32 + value_offset]), )); list_of(values, total, 4) }; ( list_of(build(0), rows, elems_per_row), - list_of(build(total as i32), rows, elems_per_row), + list_of(build(offset), rows, elems_per_row), ) } @@ -149,16 +149,21 @@ fn criterion_benchmark(c: &mut Criterion) { b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) }); - let (left, right) = nested_int_lists(rows, 8); + let (left, right) = nested_int_lists(rows, 8, (rows * 8) as i32); c.bench_function("spark_arrays_overlap: nested int32 short lists", |b| { b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) }); - let (left, right) = nested_int_lists(64, 64); + let (left, right) = nested_int_lists(64, 64, 64 * 64); c.bench_function("spark_arrays_overlap: nested int32 long lists", |b| { b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) }); + let (left, right) = nested_int_lists(rows, 8, 4); + c.bench_function("spark_arrays_overlap: nested int32 early match", |b| { + b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) + }); + let (left, right) = struct_lists(rows, 8); c.bench_function("spark_arrays_overlap: nested struct short lists", |b| { b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) diff --git a/native/spark-expr/src/array_funcs/arrays_overlap.rs b/native/spark-expr/src/array_funcs/arrays_overlap.rs index 5500a71254f..30e53bf7845 100644 --- a/native/spark-expr/src/array_funcs/arrays_overlap.rs +++ b/native/spark-expr/src/array_funcs/arrays_overlap.rs @@ -221,6 +221,18 @@ fn arrays_overlap_list( left_values.as_string::(), right_values.as_string::() ), + dt if needs_comparator(dt) => { + let comparator = make_comparator( + left_values.as_ref(), + right_values.as_ref(), + SortOptions::default(), + )?; + Ok(overlap_rows( + left, + right, + nested_row_overlap(left_values, right_values, comparator.as_ref()), + )) + } _ => arrays_overlap_list_generic(left, right), } } @@ -395,22 +407,14 @@ fn nested_row_overlap<'a>( comparator: &'a dyn Fn(usize, usize) -> Ordering, ) -> impl FnMut(Range, Range) -> bool + 'a { move |left_range, right_range| { - let (probe, probe_range, search, search_range, probe_is_left) = - if left_range.len() <= right_range.len() { - (left, left_range, right, right_range, true) - } else { - (right, right_range, left, left_range, false) - }; - - for pi in probe_range { - if probe.is_null(pi) { + for li in left_range { + if left.is_null(li) { continue; } - for si in search_range.clone() { - if search.is_null(si) { + for ri in right_range.clone() { + if right.is_null(ri) { continue; } - let (li, ri) = if probe_is_left { (pi, si) } else { (si, pi) }; if comparator(li, ri) == Ordering::Equal { return true; } @@ -420,28 +424,11 @@ fn nested_row_overlap<'a>( } } -/// Fallback for nested and otherwise unhandled element types. +/// Fallback for otherwise unhandled element types. fn arrays_overlap_list_generic( left: &GenericListArray, right: &GenericListArray, ) -> Result { - let left_values = left.values(); - let right_values = right.values(); - if left_values.data_type() == right_values.data_type() - && needs_comparator(left_values.data_type()) - { - let comparator = make_comparator( - left_values.as_ref(), - right_values.as_ref(), - SortOptions::default(), - )?; - return Ok(overlap_rows( - left, - right, - nested_row_overlap(left_values, right_values, comparator.as_ref()), - )); - } - let len = left.len(); let mut builder = BooleanArray::builder(len); @@ -739,24 +726,43 @@ mod tests { } #[test] - fn test_nested_array_sliced_offsets_and_probe_swap() -> Result<()> { - let make_rows = |rows: &[&[&[i32]]]| { + fn test_nested_array_sliced_offsets_and_nulls() -> Result<()> { + let make_rows = |rows: &[&[Option<&[i32]>]]| { let mut builder = ListBuilder::new(ListBuilder::new(Int32Builder::new())); for row in rows { for element in *row { - builder.values().values().append_slice(element); - builder.values().append(true); + if let Some(values) = element { + builder.values().values().append_slice(values); + builder.values().append(true); + } else { + builder.values().append(false); + } } builder.append(true); } builder.finish() }; - let left = make_rows(&[&[&[999]], &[&[10]], &[&[50], &[60], &[70]]]).slice(1, 2); - let right = make_rows(&[&[&[999]], &[&[20], &[30], &[40]], &[&[60]]]).slice(1, 2); + let left = make_rows(&[ + &[Some(&[999])], + &[Some(&[10])], + &[Some(&[10]), None], + &[Some(&[50]), Some(&[60]), Some(&[70])], + ]) + .slice(1, 3); + let right = make_rows(&[ + &[Some(&[999])], + &[Some(&[20]), Some(&[30]), Some(&[40])], + &[Some(&[20])], + &[Some(&[60])], + ]) + .slice(1, 3); let result = arrays_overlap_list::(&left, &right)?; let result = result.as_any().downcast_ref::().unwrap(); - assert_eq!(result, &BooleanArray::from(vec![false, true])); + assert_eq!( + result, + &BooleanArray::from(vec![Some(false), None, Some(true)]) + ); Ok(()) } 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 748b1cee231..9d201a69659 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -240,7 +240,18 @@ object CometArrayMin extends CometExpressionSerde[ArrayMin] { } } -object CometArraysOverlap extends CometExpressionSerde[ArraysOverlap] { +object CometArraysOverlap extends CometExpressionSerde[ArraysOverlap] with ArraysBase { + + override def getIncompatibleReasons(): Seq[String] = Seq(nestedFloatIncompatibilityReason) + + override def getSupportLevel(expr: ArraysOverlap): SupportLevel = { + if (hasNestedFloatElements(expr.left.dataType)) { + Incompatible(Some(nestedFloatIncompatibilityReason)) + } else { + Compatible() + } + } + override def convert( expr: ArraysOverlap, inputs: Seq[Attribute], @@ -730,12 +741,18 @@ object CometSize extends CometExpressionSerde[Size] { object CometArrayPosition extends CometExpressionSerde[ArrayPosition] with ArraysBase { + override def getIncompatibleReasons(): Seq[String] = Seq(nestedFloatIncompatibilityReason) + override def getSupportLevel(expr: ArrayPosition): SupportLevel = { if (expr.children.forall(_.foldable)) { // Fall back to Spark for all-literal args so ConstantFolding can handle it. Unsupported(Some("all arguments are literals, falling back to Spark")) } else { - childTypesSupportLevel(expr) + childTypesSupportLevel(expr) match { + case _: Compatible if hasNestedFloatElements(expr.left.dataType) => + Incompatible(Some(nestedFloatIncompatibilityReason)) + case level => level + } } } @@ -825,6 +842,18 @@ object CometArraysZip extends CometExpressionSerde[ArraysZip] { trait ArraysBase { + protected val nestedFloatIncompatibilityReason: String = + "Nested floating-point elements distinguish `-0.0` from `0.0`, unlike Spark " + + "(https://github.com/apache/datafusion-comet/issues/5191)" + + protected def hasNestedFloatElements(dt: DataType): Boolean = dt match { + case ArrayType(elementType: ArrayType, _) => + SupportLevel.containsType(elementType, classOf[FloatType], classOf[DoubleType]) + case ArrayType(elementType: StructType, _) => + SupportLevel.containsType(elementType, classOf[FloatType], classOf[DoubleType]) + case _ => false + } + def isTypeSupported(dt: DataType): Boolean = { import DataTypes._ dt match { diff --git a/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql b/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql index f2a47a33ac1..5d3283d61d0 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql @@ -182,6 +182,16 @@ INSERT INTO test_overlap_nested VALUES (array(array(1, 2), array(3, 4)), array(a query SELECT a, b, arrays_overlap(a, b) FROM test_overlap_nested +-- nested floating-point signed-zero behavior differs from Spark +statement +CREATE TABLE test_overlap_nested_dbl(a array>, b array>) USING parquet + +statement +INSERT INTO test_overlap_nested_dbl VALUES (array(array(0.0D)), array(array(-0.0D))), (array(array(double('NaN'))), array(array(double('NaN')))) + +query ignore(https://github.com/apache/datafusion-comet/issues/5191) +SELECT a, b, arrays_overlap(a, b) FROM test_overlap_nested_dbl + -- struct element arrays statement CREATE TABLE test_overlap_struct(a array>, b array>) USING parquet From 390fc91254e1877f12165ac858042f980d3860b7 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Fri, 28 Aug 2026 00:23:29 +0800 Subject: [PATCH 3/5] docs: clarify nested-float fallback wording and map unreachability - Comment in hasNestedFloatElements that map elements are rejected by Spark's analyzer (TypeUtils.checkForOrderingExpr) before planning - Reword arrays_overlap/array_position notes in expressions.md to state the nested-float case falls back to Spark by default, with the native path opt-in via allowIncompatible Co-Authored-By: Claude Fable 5 --- docs/source/user-guide/latest/expressions.md | 4 ++-- spark/src/main/scala/org/apache/comet/serde/arrays.scala | 3 +++ 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index ea1af859035..571d77116a9 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -158,12 +158,12 @@ The tables below list every Spark built-in expression with its current status. | `array_join` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the incompatible native path is opt-in via allowIncompatible ([details](compatibility/expressions/array.md)) | | `array_max` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | | `array_min` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | -| `array_position` | ✅ | Native | Binary/struct/map/null elements fall back; nested floating-point signed-zero handling differs ([#5191](https://github.com/apache/datafusion-comet/issues/5191)) | +| `array_position` | ✅ | Native | Binary/struct/map/null elements fall back; arrays with nested floating-point elements fall back to Spark by default, and the incompatible native path is opt-in via allowIncompatible ([#5191](https://github.com/apache/datafusion-comet/issues/5191)) | | `array_prepend` | ✅ | — | | | `array_remove` | ✅ | Native | | | `array_repeat` | ✅ | Native | | | `array_union` | ✅ | Native | NaN/signed-zero handling may differ ([details](compatibility/floating-point.md)) | -| `arrays_overlap` | ✅ | Native | Nested floating-point signed-zero handling differs ([#5191](https://github.com/apache/datafusion-comet/issues/5191)) | +| `arrays_overlap` | ✅ | Native | Arrays with nested floating-point elements fall back to Spark by default, and the incompatible native path is opt-in via allowIncompatible ([#5191](https://github.com/apache/datafusion-comet/issues/5191)) | | `arrays_zip` | ✅ | Native | | | `element_at` | ✅ | Native | | | `flatten` | ✅ | Native | Binary/struct/map elements fall back | 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 9d201a69659..886536209e7 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -851,6 +851,9 @@ trait ArraysBase { SupportLevel.containsType(elementType, classOf[FloatType], classOf[DoubleType]) case ArrayType(elementType: StructType, _) => SupportLevel.containsType(elementType, classOf[FloatType], classOf[DoubleType]) + // Map elements cannot reach here: ArraysOverlap and ArrayPosition call + // TypeUtils.checkForOrderingExpr on the element type, and maps are not orderable, + // so Spark's analyzer rejects them before planning. case _ => false } From 619fc4252752d89179c4d1418c58ab4a06f18ca4 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Tue, 29 Sep 2026 01:51:07 +0800 Subject: [PATCH 4/5] fix: keep nested arrays_overlap types that differ in nullability on the comparator The strict element-type equality check sent nested types that differ only in field nullability, such as a struct from array_repeat beside one from array(), to the generic fallback, whose flat equality kernel rejects nested types. Keep every nested pair on the shared make_comparator path, which compares them as the pre-PR per-row comparator did; only flat types need identical types for the fast paths. --- .../src/array_funcs/arrays_overlap.rs | 39 ++++++++++++++++++- .../expressions/array/arrays_overlap.sql | 12 ++++++ 2 files changed, 50 insertions(+), 1 deletion(-) diff --git a/native/spark-expr/src/array_funcs/arrays_overlap.rs b/native/spark-expr/src/array_funcs/arrays_overlap.rs index e23a010b634..1799da467a2 100644 --- a/native/spark-expr/src/array_funcs/arrays_overlap.rs +++ b/native/spark-expr/src/array_funcs/arrays_overlap.rs @@ -170,7 +170,13 @@ fn arrays_overlap_list( let left_values = left.values(); let right_values = right.values(); - if left_values.data_type() != right_values.data_type() { + // Nested element types can differ in field nullability, such as a struct built by + // `array_repeat` beside one built by `array(...)`. `make_comparator` compares those, so keep + // them on the shared comparator below; only flat types need identical types for the fast + // paths and otherwise take the generic fallback. + let both_nested = + needs_comparator(left_values.data_type()) && needs_comparator(right_values.data_type()); + if left_values.data_type() != right_values.data_type() && !both_nested { return arrays_overlap_list_generic(left, right); } @@ -1082,6 +1088,37 @@ mod tests { list_builder.finish() } + #[test] + fn test_struct_overlap_with_different_field_nullability() -> Result<()> { + // The same struct values where one side declares `a` non-nullable, as `array_repeat` + // produces while `array(...)` widens it to nullable: [{1,2}] vs [{1,2}] => true + fn single_struct_list(a_nullable: bool) -> ListArray { + let fields = vec![ + Arc::new(Field::new("a", DataType::Int32, a_nullable)), + Arc::new(Field::new("b", DataType::Int32, true)), + ]; + let mut list_builder = ListBuilder::new(StructBuilder::new( + fields, + vec![Box::new(Int32Builder::new()), Box::new(Int32Builder::new())], + )); + let sb = list_builder.values(); + sb.field_builder::(0).unwrap().append_value(1); + sb.field_builder::(1).unwrap().append_value(2); + sb.append(true); + list_builder.append(true); + list_builder.finish() + } + let left = single_struct_list(false); + let right = single_struct_list(true); + assert_ne!(left.values().data_type(), right.values().data_type()); + + let result = arrays_overlap_list::(&left, &right)?; + let result = result.as_any().downcast_ref::().unwrap(); + assert!(result.is_valid(0)); + assert!(result.value(0)); + Ok(()) + } + #[test] fn test_struct_basic_overlap() -> Result<()> { // [{1,2}, {3,4}] vs [{3,4}, {5,6}] => true diff --git a/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql b/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql index 1ddfe14dbcd..5ca5db839b9 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/arrays_overlap.sql @@ -275,6 +275,18 @@ INSERT INTO test_overlap_struct VALUES (array(named_struct('x', 1, 'y', 2)), arr query SELECT a, b, arrays_overlap(a, b) FROM test_overlap_struct +-- array_repeat keeps the non-nullable x field while array() widens it to nullable, so the two +-- sides' element types differ only in nested nullability +statement +CREATE TABLE test_overlap_mixed_ctor(i int) USING parquet + +statement +INSERT INTO test_overlap_mixed_ctor VALUES (1), (2), (NULL) + +query +SELECT i, arrays_overlap(array_repeat(named_struct('x', 1, 'y', i), 1), array(named_struct('x', 1, 'y', i))) +FROM test_overlap_mixed_ctor + -- mixed column and literal with NULL elements query SELECT arrays_overlap(a, array(99, NULL)) FROM test_arrays_overlap From 204871f0d2c1cdab7b742e707f5d6d5332c0eaba Mon Sep 17 00:00:00 2001 From: peterxcli Date: Sun, 4 Oct 2026 23:52:07 +0800 Subject: [PATCH 5/5] perf: probe nested arrays_overlap rows from the shorter side The hoisted comparator scanned every nested row left side first. When the match is late in the longer side and early in the shorter side, that scan does close to n*m comparisons, where the per-row comparator on main did n. For a = [[0], ..., [127]] and b = [[127], ..., [190]], the count went from 128 to 8,129. Branch once per row, so that the shorter side is the outer loop again, as on main. The comparator still takes (left_index, right_index), because it compares left against right. A unit test counts the comparator calls in both argument orders. The benchmark adds two unequal-length shapes: the match last in the longer side, and the match last in the shorter side. --- native/spark-expr/benches/arrays_overlap.rs | 39 ++++++++++++ .../src/array_funcs/arrays_overlap.rs | 62 ++++++++++++++++--- 2 files changed, 91 insertions(+), 10 deletions(-) diff --git a/native/spark-expr/benches/arrays_overlap.rs b/native/spark-expr/benches/arrays_overlap.rs index 67f5b56d138..49a19795e5f 100644 --- a/native/spark-expr/benches/arrays_overlap.rs +++ b/native/spark-expr/benches/arrays_overlap.rs @@ -23,6 +23,7 @@ use datafusion::common::config::ConfigOptions; use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl}; use datafusion_comet_spark_expr::SparkArraysOverlap; use std::hint::black_box; +use std::ops::Range; use std::sync::Arc; fn list_of(values: ArrayRef, rows: usize, elems_per_row: usize) -> ArrayRef { @@ -85,6 +86,30 @@ fn nested_int_lists(rows: usize, elems_per_row: usize, offset: i32) -> (ArrayRef ) } +/// Nested int32 lists of one-value lists, none of them null. Each left row holds `[v]` for each +/// `v` in `left`, and each right row likewise for `right`. +fn nested_singleton_lists( + rows: usize, + left: Range, + right: Range, +) -> (ArrayRef, ArrayRef) { + let build = |values: Range| { + let len = values.len(); + let inner: ArrayRef = Arc::new(Int32Array::from_iter_values( + (0..rows).flat_map(|_| values.clone()), + )); + let offsets: Vec = (0..=rows * len).map(|i| i as i32).collect(); + let singletons: ArrayRef = Arc::new(ListArray::new( + Arc::new(Field::new("item", DataType::Int32, true)), + OffsetBuffer::new(offsets.into()), + inner, + None, + )); + list_of(singletons, rows, len) + }; + (build(left), build(right)) +} + fn struct_lists(rows: usize, elems_per_row: usize) -> (ArrayRef, ArrayRef) { let total = rows * elems_per_row; let build = |offset: i32| -> ArrayRef { @@ -164,6 +189,20 @@ fn criterion_benchmark(c: &mut Criterion) { b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) }); + // Rows hold 128 and 64 elements with one match. The first bench puts it last in the longer + // side, the second last in the shorter side. + let (left, right) = nested_singleton_lists(256, 0..128, 127..191); + c.bench_function( + "spark_arrays_overlap: nested int32 unequal lengths, late match in longer side", + |b| b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))), + ); + + let (left, right) = nested_singleton_lists(256, 0..128, -63..1); + c.bench_function( + "spark_arrays_overlap: nested int32 unequal lengths, late match in shorter side", + |b| b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))), + ); + let (left, right) = struct_lists(rows, 8); c.bench_function("spark_arrays_overlap: nested struct short lists", |b| { b.iter(|| black_box(invoke(&udf, black_box(&left), black_box(&right)))) diff --git a/native/spark-expr/src/array_funcs/arrays_overlap.rs b/native/spark-expr/src/array_funcs/arrays_overlap.rs index 1799da467a2..fa5410fb98e 100644 --- a/native/spark-expr/src/array_funcs/arrays_overlap.rs +++ b/native/spark-expr/src/array_funcs/arrays_overlap.rs @@ -433,21 +433,41 @@ fn nested_row_overlap<'a>( comparator: &'a dyn Fn(usize, usize) -> Ordering, ) -> impl FnMut(Range, Range) -> bool + 'a { move |left_range, right_range| { - for li in left_range { - if left.is_null(li) { + // Probe from the shorter side, as the per-row comparator did, so its early exits stay. + // The comparator takes the left index first, also when the right side is the outer loop. + if left_range.len() <= right_range.len() { + any_equal(left, left_range, right, right_range, comparator) + } else { + any_equal(right, right_range, left, left_range, |ri, li| { + comparator(li, ri) + }) + } + } +} + +/// True when a non-null `outer` element equals a non-null `inner` element. `compare` takes an +/// `outer` index, then an `inner` index. +fn any_equal( + outer: &ArrayRef, + outer_range: Range, + inner: &ArrayRef, + inner_range: Range, + compare: impl Fn(usize, usize) -> Ordering, +) -> bool { + for o in outer_range { + if outer.is_null(o) { + continue; + } + for i in inner_range.clone() { + if inner.is_null(i) { continue; } - for ri in right_range.clone() { - if right.is_null(ri) { - continue; - } - if comparator(li, ri) == Ordering::Equal { - return true; - } + if compare(o, i) == Ordering::Equal { + return true; } } - false } + false } fn normalize_list_element_floats( @@ -916,6 +936,28 @@ mod tests { Ok(()) } + #[test] + fn test_nested_scan_probes_from_the_shorter_side() { + // The match is the last element of the longer side and the first of the shorter side. + // A reversed argument order reads past the shorter side and panics. + let long: ArrayRef = Arc::new(Int32Array::from_iter_values(0..128)); + let short: ArrayRef = Arc::new(Int32Array::from_iter_values(127..191)); + for (left, right) in [(&long, &short), (&short, &long)] { + let calls = std::cell::Cell::new(0); + let (left_values, right_values) = ( + left.as_primitive::(), + right.as_primitive::(), + ); + let comparator = |li: usize, ri: usize| { + calls.set(calls.get() + 1); + left_values.value(li).cmp(&right_values.value(ri)) + }; + let mut row_overlap = nested_row_overlap(left, right, &comparator); + assert!(row_overlap(0..left.len(), 0..right.len())); + assert_eq!(calls.get(), 128); + } + } + #[test] fn test_nested_array_sliced_offsets_and_nulls() -> Result<()> { let make_rows = |rows: &[&[Option<&[i32]>]]| {