Skip to content
Open
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
62 changes: 61 additions & 1 deletion native/spark-expr/benches/arrays_overlap.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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, offset: i32) -> (ArrayRef, ArrayRef) {
let total = rows * elems_per_row;
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 + value_offset]),
));
list_of(values, total, 4)
};
(
list_of(build(0), rows, elems_per_row),
list_of(build(offset), 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![
Expand Down Expand Up @@ -113,6 +148,31 @@ 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, (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, 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))))
});

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);
Expand Down
167 changes: 130 additions & 37 deletions native/spark-expr/src/array_funcs/arrays_overlap.rs
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,13 @@ fn arrays_overlap_list<OffsetSize: OffsetSizeTrait>(
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);
}

Expand Down Expand Up @@ -223,6 +229,29 @@ fn arrays_overlap_list<OffsetSize: OffsetSizeTrait>(
left_values.as_string::<i64>(),
right_values.as_string::<i64>()
),
dt if needs_comparator(dt) => {
// Spark's nested path compares with ordering.equiv, where -0.0 == 0.0 and every NaN
// is equal, but Arrow's comparator uses total order. Normalize float leaves once per
// column so the comparator built over the full child arrays matches Spark.
let (left_values, right_values) = if has_float_leaf(dt) {
(
normalize_nested_floats(left_values),
normalize_nested_floats(right_values),
)
} else {
(Arc::clone(left_values), Arc::clone(right_values))
};
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),
}
}
Expand Down Expand Up @@ -397,6 +426,30 @@ 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<usize>, Range<usize>) -> bool + 'a {
move |left_range, right_range| {
for li in left_range {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Could we preserve shorter-side probing while hoisting the comparator? For arrays_overlap(a, b) on nested integer columns where a=[[0],...,[127]] and b=[[127],...,[190]], both versions correctly return true, but the base probes the shorter right side and finds the match after 128 comparisons. This unconditional left-first loop performs 8,129 comparisons. Two independent release-mode measurements reproduced approximately 50× higher kernel latency, so this materially slows previously efficient overlapping-array workloads. Branch once per row to preserve the previous probe order, keeping comparator arguments in (left_index, right_index) order, and cover unequal lengths in the benchmark.

Evidence: Verified the extracted base/head kernel bodies against the requested Git commits. With locked Arrow 59.3.0, 128 rows, and every twentieth outer row null, alternating base/head measurements produced 0.148 ms versus 7.380 ms and 0.148 ms versus 7.406 ms. Outputs matched. A separately instrumented single-row probe counted exactly 128 versus 8,129 comparator calls. Reproduce with cargo run --release --offline --locked --manifest-path /tmp/comet-5194-b641-kernel/Cargo.toml --bin comet-5194-b641-kernel and --bin count. Timing logs are /tmp/comet-5194-final-bench1.log and /tmp/comet-5194-final-bench2.log.

if left.is_null(li) {
continue;
}
for ri in right_range.clone() {
if right.is_null(ri) {
continue;
}
if comparator(li, ri) == Ordering::Equal {
return true;
}
}
}
false
}
}

fn normalize_list_element_floats<OffsetSize: OffsetSizeTrait>(
list: &GenericListArray<OffsetSize>,
) -> GenericListArray<OffsetSize> {
Expand All @@ -413,10 +466,11 @@ fn normalize_list_element_floats<OffsetSize: OffsetSizeTrait>(
)
}

/// Fallback for nested and otherwise unhandled element types.
/// Fallback for otherwise unhandled element types, including nested elements whose two sides
/// have different data 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
/// only the nested path treats them as equal. this normalization can't move into the
/// flat fast path in arrays_overlap_list without breaking that difference.
fn arrays_overlap_list_generic<OffsetSize: OffsetSizeTrait>(
left: &GenericListArray<OffsetSize>,
Expand Down Expand Up @@ -464,26 +518,12 @@ fn arrays_overlap_list_generic<OffsetSize: OffsetSizeTrait>(
(&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)?;

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Could we preserve nested comparator handling in this fallback? The caller still routes unequal Arrow child types here, including types that differ only in nullability. For a Parquet table t(i INT), SELECT arrays_overlap(array_repeat(named_struct('x',1,'y',i),1), array(named_struct('x',1,'y',i))) FROM t should return true. array_repeat preserves the non-nullable x field, while CometCreateArray widens it to nullable. The previous make_comparator handled this difference, but the unconditional find_in_array_flat now raises Nested comparison ... (hint: use make_comparator instead), aborting a previously working query. Keep comparator support in the fallback or dispatch compatible nested types before the strict metadata-equality guard, and cover this mixed-constructor case.

Evidence: Using exact base and head implementations with locked Arrow 59.3.0, disposable Rust probes compared identical List and Struct values whose child nullability differed. Both returned true on the base and Arrow errors on the head. A third probe using CreateNamedStruct, Spark's native array_repeat, Comet's spark_cast, and make_array produced the SQL input types and reproduced the same regression. Spark 4.1.3 returned [true, true, true] for i = 1, 2, NULL. Reproduction source and output are retained at /tmp/comet-5194-reproduction.rs and /tmp/comet-5194-repro.log.

if null_eq {
has_null = true;
}
Expand Down Expand Up @@ -513,25 +553,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,
Expand Down Expand Up @@ -895,6 +916,47 @@ mod tests {
Ok(())
}

#[test]
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 {
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(&[
&[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::<i32>(&left, &right)?;
let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
assert_eq!(
result,
&BooleanArray::from(vec![Some(false), None, Some(true)])
);
Ok(())
}

#[test]
fn test_nested_array_basic_overlap() -> Result<()> {
// [[1,2], [3,4]] vs [[3,4], [5,6]] => true
Expand Down Expand Up @@ -1026,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::<Int32Builder>(0).unwrap().append_value(1);
sb.field_builder::<Int32Builder>(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::<i32>(&left, &right)?;
let result = result.as_any().downcast_ref::<BooleanArray>().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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading