diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index 5cdb222eb1..cd167885d0 100644 --- a/docs/source/contributor-guide/expression-audits/array_funcs.md +++ b/docs/source/contributor-guide/expression-audits/array_funcs.md @@ -91,19 +91,19 @@ ## array_max -- Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8. -- Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayMax(child) extends UnaryExpression with ImplicitCastInputTypes`; skips NULL elements; for float/double Spark's `SQLOrderingUtil` treats NaN as greater than any non-NaN. Wired as `CometScalarFunction("array_max")`. -- 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. -- Float/double arrays containing NaN match Spark: NaN is treated as greater than any non-NaN value. +- Spark 3.4.3 (audited 2026-08-22): identical to 3.5.8. +- Spark 3.5.8 (audited 2026-08-22): `ArrayMax` skips NULL elements and returns NULL for an empty or all-NULL array. `SQLOrderingUtil` treats all NaNs as equal and greater than non-NaN values, and signed zeros as equal. The first equal maximum is retained. Nested arrays and structs compare lexicographically, with NULL fields or elements ordered first. +- Spark 4.0.1 (audited 2026-08-22): `NullIntolerant` becomes a `nullIntolerant` field. Extrema semantics are unchanged; string ordering can use non-default collations. +- Spark 4.1.1 (audited 2026-08-22): identical to 4.0.1. +- Current status: `CometArrayMax` uses the native `SparkArrayExtrema` UDF. Typed float/double scans and recursive array/struct comparisons follow Spark's ordering and preserve the original first equal element, including its zero sign and NaN representation. This path is used in both strict and non-strict floating-point modes without the JVM codegen dispatcher. Other scalar element types retain the existing DataFusion implementation. Non-UTF8_BINARY string collations, including nested fields, use Spark's JVM codegen dispatcher inside the Comet pipeline by default. If the dispatcher is disabled, these cases fall back to Spark unless incompatible native execution is explicitly enabled ([#4496](https://github.com/apache/datafusion-comet/issues/4496)). ## array_min -- Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8. -- Spark 3.5.8 (audited 2026-05-27): mirror of `ArrayMax` with `evalInternal` returning the minimum. Same NULL-skip and NaN-ordering semantics. Wired as `CometScalarFunction("array_min")`. -- Spark 4.0.1 (audited 2026-05-27): same trait refactor as `array_max`. -- Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1. -- Float/double arrays containing NaN match Spark, mirroring `array_max`. +- Spark 3.4.3 (audited 2026-08-22): identical to 3.5.8. +- Spark 3.5.8 (audited 2026-08-22): mirrors `ArrayMax`, retaining the first equal minimum. The NULL, NaN, signed-zero, and nested comparison rules are the same. +- Spark 4.0.1 (audited 2026-08-22): same trait refactor and collation support as `array_max`, with no change in floating-point extrema semantics. +- Spark 4.1.1 (audited 2026-08-22): identical to 4.0.1. +- Current status: `CometArrayMin` shares the native `SparkArrayExtrema` implementation and support boundary with `array_max`. Both floating-point modes use Spark-compatible native ordering, preserving the original first equal minimum. Non-default string collations use the same JVM codegen dispatch and dispatcher-disabled fallback as `array_max` ([#4496](https://github.com/apache/datafusion-comet/issues/4496)). ## array_position diff --git a/docs/source/user-guide/latest/compatibility/floating-point.md b/docs/source/user-guide/latest/compatibility/floating-point.md index b39ef7f645..f7e32d9f85 100644 --- a/docs/source/user-guide/latest/compatibility/floating-point.md +++ b/docs/source/user-guide/latest/compatibility/floating-point.md @@ -28,6 +28,13 @@ to Spark in some cases, especially when the data contains both positive and nega case that is not of concern for many users. If it is a concern, setting `spark.comet.exec.strictFloatingPoint=true` will make relevant operations fall back to Spark. +`array_min` and `array_max` use Spark-compatible native comparisons in both strict and non-strict +floating-point modes. Signed zeros compare equal, and all NaN representations compare equal and +greater than non-NaN values. The original first equal element is retained: for example, +`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. + ## Ordering: NaN and signed zero (`-0.0` vs `+0.0`) Spark's `ORDER BY`, `RANK`, `DENSE_RANK`, and window frame comparisons route through diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 501ce468c4..b056bed5d6 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -156,8 +156,8 @@ The tables below list every Spark built-in expression with its current status. | `array_insert` | ✅ | Native | | | `array_intersect` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the incompatible native path is opt-in via allowIncompatible ([details](compatibility/expressions/array.md)) | | `array_join` | ✅ | Hybrid | Native for literal or column delimiter and null replacement; other cases and non-UTF8_BINARY collations use the JVM codegen dispatcher ([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_max` | ✅ | Hybrid | Native Spark-compatible floating-point and nested ordering; non-default string collations use the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | +| `array_min` | ✅ | Hybrid | Native Spark-compatible floating-point and nested ordering; non-default string collations use the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | | `array_position` | ✅ | Native | Binary/struct/map/null elements fall back | | `array_prepend` | ✅ | — | | | `array_remove` | ✅ | Native | | diff --git a/native/spark-expr/Cargo.toml b/native/spark-expr/Cargo.toml index 380656f107..f225fe6b2d 100644 --- a/native/spark-expr/Cargo.toml +++ b/native/spark-expr/Cargo.toml @@ -259,6 +259,10 @@ harness = false name = "arrays_overlap" harness = false +[[bench]] +name = "array_extrema" +harness = false + [[bench]] name = "checked_arithmetic" harness = false diff --git a/native/spark-expr/benches/array_extrema.rs b/native/spark-expr/benches/array_extrema.rs new file mode 100644 index 0000000000..ea52946fed --- /dev/null +++ b/native/spark-expr/benches/array_extrema.rs @@ -0,0 +1,137 @@ +// 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 ordinary-data extrema with DataFusion; special-value semantics belong in tests. + +use arrow::array::{ArrayRef, Float32Array, Float64Array, ListArray}; +use arrow::buffer::{NullBuffer, OffsetBuffer}; +use arrow::datatypes::Field; +use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion}; +use datafusion::common::config::ConfigOptions; +use datafusion::functions_nested::min_max::{array_max_udf, array_min_udf}; +use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDF}; +use datafusion_comet_spark_expr::SparkArrayExtrema; +use std::hint::black_box; +use std::sync::Arc; +use std::time::Duration; + +fn list(values: ArrayRef, len: usize, nullable: bool) -> ArrayRef { + let rows = values.len() / len; + Arc::new(ListArray::new( + Arc::new(Field::new_list_field(values.data_type().clone(), true)), + OffsetBuffer::from_lengths(std::iter::repeat_n(len, rows)), + values, + nullable.then(|| NullBuffer::from_iter((0..rows).map(|i| i % 10 != 0))), + )) +} + +fn criterion_benchmark(c: &mut Criterion) { + let mut group = c.benchmark_group("array_extrema"); + group.sample_size(20); + group.warm_up_time(Duration::from_millis(250)); + group.measurement_time(Duration::from_secs(1)); + for len in [8, 1024] { + for nullable in [false, true] { + let values = (0..64 * len) + .map(|i| (!nullable || i % 10 != 0).then_some(((i * 17) % 1000 + 1) as f64)); + let inputs = [ + ( + "float32", + list( + Arc::new(Float32Array::from_iter( + values.clone().map(|v| v.map(|v| v as f32)), + )), + len, + nullable, + ), + ), + ( + "float64", + list(Arc::new(Float64Array::from_iter(values)), len, nullable), + ), + ( + "nested", + // Null list elements are skipped by both engines. Keep inner floats non-null + // so the fixture does not depend on their different nested null ordering. + list( + list( + Arc::new(Float64Array::from_iter_values( + (0..64 * len * 4).map(|i| ((i * 17) % 1000 + 1) as f64), + )), + 4, + nullable, + ), + len, + nullable, + ), + ), + ]; + for (kind, input) in inputs { + for is_min in [true, false] { + let comet = ScalarUDF::from(SparkArrayExtrema::new(is_min)); + let datafusion = if is_min { + array_min_udf() + } else { + array_max_udf() + }; + let args = ScalarFunctionArgs { + args: vec![ColumnarValue::Array(Arc::clone(&input))], + arg_fields: vec![Arc::new(Field::new( + "input", + input.data_type().clone(), + true, + ))], + number_rows: input.len(), + return_field: Arc::new(Field::new( + "result", + comet.return_type(&[input.data_type().clone()]).unwrap(), + true, + )), + config_options: Arc::new(ConfigOptions::default()), + }; + let evaluate = |udf: &ScalarUDF| { + udf.invoke_with_args(args.clone()) + .unwrap() + .into_array(input.len()) + .unwrap() + }; + assert_eq!(evaluate(&comet).to_data(), evaluate(&datafusion).to_data()); + let op = if is_min { "min" } else { "max" }; + for (engine, udf) in [("comet", &comet), ("datafusion", datafusion.as_ref())] { + group.bench_function( + BenchmarkId::new( + format!("{kind}_{op}_{engine}"), + format!("len={len}_null={nullable}"), + ), + |b| { + b.iter(|| { + black_box( + udf.invoke_with_args(black_box(args.clone())).unwrap(), + ) + }) + }, + ); + } + } + } + } + } + group.finish(); +} + +criterion_group!(benches, criterion_benchmark); +criterion_main!(benches); diff --git a/native/spark-expr/src/array_funcs/array_extrema.rs b/native/spark-expr/src/array_funcs/array_extrema.rs new file mode 100644 index 0000000000..b2638b6648 --- /dev/null +++ b/native/spark-expr/src/array_funcs/array_extrema.rs @@ -0,0 +1,322 @@ +// 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. + +use std::cmp::Ordering; +use std::sync::Arc; + +use arrow::array::{ + make_array, make_comparator, new_empty_array, Array, ArrayRef, AsArray, DynComparator, + ListArray, MutableArrayData, PrimitiveArray, PrimitiveBuilder, StructArray, UInt32Array, +}; +use arrow::buffer::NullBuffer; +use arrow::compute::{take, SortOptions}; +use arrow::datatypes::{ArrowPrimitiveType, DataType, Float32Type, Float64Type}; +use datafusion::common::{exec_err, Result, ScalarValue}; +use datafusion::functions_nested::min_max::{array_max_udf, array_min_udf}; +use datafusion::logical_expr::{ + ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, +}; +use num::Float; + +/// Spark's array_min/array_max retain the first non-null value on an ordering tie. +/// In particular, signed zeros compare equal and all NaNs compare equal and greater +/// than non-NaNs. Nested arrays and structs use the same ordering, with nulls first. +#[derive(Debug, Hash, Eq, PartialEq)] +pub struct SparkArrayExtrema { + is_min: bool, + datafusion_udf: Arc, +} + +impl SparkArrayExtrema { + pub fn new(is_min: bool) -> Self { + Self { + is_min, + // Capture the original implementation, not a registry lookup: these UDFs + // replace the DataFusion names in Comet's function registry. + datafusion_udf: if is_min { + array_min_udf() + } else { + array_max_udf() + }, + } + } +} + +impl ScalarUDFImpl for SparkArrayExtrema { + fn name(&self) -> &str { + if self.is_min { + "array_min" + } else { + "array_max" + } + } + + fn signature(&self) -> &Signature { + self.datafusion_udf.signature() + } + + fn return_type(&self, arg_types: &[DataType]) -> Result { + self.datafusion_udf.return_type(arg_types) + } + + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { + let [input] = args.args.as_slice() else { + return exec_err!("{} takes exactly one argument", self.name()); + }; + let element_type = self.return_type(&[input.data_type()])?; + + // DataFusion's non-primitive path reconstructs an array from scalars, which + // cannot infer a type from an empty iterator. Keep the declared element type. + if matches!(input, ColumnarValue::Array(array) if array.is_empty()) { + return Ok(ColumnarValue::Array(new_empty_array(&element_type))); + } + if !matches!( + element_type, + DataType::Float32 | DataType::Float64 | DataType::List(_) | DataType::Struct(_) + ) { + return self.datafusion_udf.invoke_with_args(args); + } + + let is_scalar = matches!(input, ColumnarValue::Scalar(_)); + let array = match input { + ColumnarValue::Array(array) => Arc::clone(array), + ColumnarValue::Scalar(value) => value.to_array()?, + }; + // Spark arrays use Arrow's 32-bit List layout. + let result = array_extrema(array.as_list::(), self.is_min)?; + + if is_scalar { + Ok(ColumnarValue::Scalar(ScalarValue::try_from_array( + &result, 0, + )?)) + } else { + Ok(ColumnarValue::Array(result)) + } + } +} + +fn array_extrema(array: &ListArray, is_min: bool) -> Result { + match array.value_type() { + DataType::Float32 => Ok(Arc::new(float_extrema::(array, is_min))), + DataType::Float64 => Ok(Arc::new(float_extrema::(array, is_min))), + _ => nested_extrema(array, is_min), + } +} + +/// Scan the flat value buffer for every list length. Arrow's float min/max kernels +/// use a different ordering, so long lists must not switch to those kernels. +/// Checking only Arrow's winner cannot establish that its answer agrees with Spark: +/// for max([-NaN, 1.0]), Arrow's total ordering selects 1.0 while Spark selects NaN. +/// A corrective scan triggered only by a zero or NaN winner would miss that case. +fn float_extrema(array: &ListArray, is_min: bool) -> PrimitiveArray +where + T::Native: Float, +{ + let values = array.values().as_primitive::(); + let buffer = values.values(); + let nulls = values.nulls(); + let mut result = PrimitiveBuilder::::with_capacity(array.len()); + for (row, offsets) in array.offsets().windows(2).enumerate() { + let mut best: Option = None; + if array.is_valid(row) { + let start = offsets[0] as usize; + let end = offsets[1] as usize; + for (index, &candidate) in buffer[start..end].iter().enumerate() { + if nulls.is_some_and(|nulls| nulls.is_null(start + index)) { + continue; + } + let replace = match best { + None => true, + Some(current) if is_min => { + candidate < current || (!candidate.is_nan() && current.is_nan()) + } + Some(current) => { + candidate > current || (candidate.is_nan() && !current.is_nan()) + } + }; + if replace { + // Copy the winning value, never normalize its zero sign or NaN bits. + best = Some(candidate); + } + } + } + result.append_option(best); + } + result.finish() +} + +fn nested_extrema(array: &ListArray, is_min: bool) -> Result { + let values = array.values(); + let compare = spark_comparator(values)?; + let nulls = values.nulls(); + let ordering = if is_min { + Ordering::Less + } else { + Ordering::Greater + }; + let mut indices = Vec::with_capacity(array.len()); + for (row, offsets) in array.offsets().windows(2).enumerate() { + let mut best = None; + if array.is_valid(row) { + for candidate in offsets[0] as usize..offsets[1] as usize { + if nulls.is_some_and(|nulls| nulls.is_null(candidate)) { + continue; + } + if best.is_none_or(|current| compare(candidate, current) == ordering) { + best = Some(candidate); + } + } + } + indices.push(best.map(|index| index as u32)); + } + take_extrema_values(values, &UInt32Array::from(indices)) +} + +fn take_extrema_values(values: &ArrayRef, indices: &UInt32Array) -> Result { + match values.data_type() { + // Arrow's flat-list take is faster, but its child capacity estimate can + // grow excessively for sparse outputs or recursively nested children. + DataType::List(field) + if indices.len() <= values.len() && !field.data_type().is_nested() => + { + let mut result = take(values.as_ref(), indices, None)?; + result.shrink_to_fit(); + Ok(result) + } + DataType::List(_) => { + // Start nested children empty, copying only the selected values. + let data = values.to_data(); + let mut result = MutableArrayData::new(vec![&data], true, 0); + for index in indices.iter() { + match index.filter(|&index| values.is_valid(index as usize)) { + Some(index) => result.try_extend(0, index as usize, index as usize + 1)?, + None => result.try_extend_nulls(1)?, + } + } + Ok(make_array(result.freeze())) + } + DataType::Struct(fields) => { + let columns = values + .as_struct() + .columns() + .iter() + .map(|column| take_extrema_values(column, indices)) + .collect::>>()?; + let nulls = indices + .iter() + .map(|index| index.is_some_and(|index| values.is_valid(index as usize))) + .collect::(); + Ok(Arc::new(StructArray::try_new_with_length( + fields.clone(), + columns, + Some(nulls), + indices.len(), + )?)) + } + _ => Ok(take(values.as_ref(), indices, None)?), + } +} + +/// Build one comparator per child array, not per row. This is local to extrema: +/// DataFusion's ScalarValue nested comparisons put inner nulls last, unlike Spark. +fn spark_comparator(array: &ArrayRef) -> Result { + match array.data_type() { + DataType::Float32 => Ok(float_comparator::(array)), + DataType::Float64 => Ok(float_comparator::(array)), + DataType::List(_) => { + let array = array.as_list::(); + let compare = spark_comparator(array.values())?; + let offsets = array.offsets().clone(); + Ok(nulls_first(array.nulls().cloned(), move |left, right| { + let left_start = offsets[left] as usize; + let right_start = offsets[right] as usize; + let left_len = offsets[left + 1] as usize - left_start; + let right_len = offsets[right + 1] as usize - right_start; + for offset in 0..left_len.min(right_len) { + let ordering = compare(left_start + offset, right_start + offset); + if ordering != Ordering::Equal { + return ordering; + } + } + left_len.cmp(&right_len) + })) + } + DataType::Struct(_) => { + let array = array.as_struct(); + let fields = array + .columns() + .iter() + .map(spark_comparator) + .collect::>>()?; + Ok(nulls_first(array.nulls().cloned(), move |left, right| { + fields + .iter() + .map(|compare| compare(left, right)) + .find(|&ordering| ordering != Ordering::Equal) + .unwrap_or(Ordering::Equal) + })) + } + _ => Ok(make_comparator( + array.as_ref(), + array.as_ref(), + SortOptions { + descending: false, + nulls_first: true, + }, + )?), + } +} + +fn float_comparator(array: &ArrayRef) -> DynComparator +where + T::Native: Float, +{ + let values = array.as_primitive::().values().clone(); + nulls_first(array.nulls().cloned(), move |left, right| { + let left = values[left]; + let right = values[right]; + if left == right || (left.is_nan() && right.is_nan()) { + Ordering::Equal + } else if left > right || left.is_nan() { + Ordering::Greater + } else { + Ordering::Less + } + }) +} + +fn nulls_first( + nulls: Option, + compare: impl Fn(usize, usize) -> Ordering + Send + Sync + 'static, +) -> DynComparator { + match nulls { + None => Box::new(compare), + Some(nulls) => { + Box::new( + move |left, right| match (nulls.is_null(left), nulls.is_null(right)) { + (true, true) => Ordering::Equal, + (true, false) => Ordering::Less, + (false, true) => Ordering::Greater, + (false, false) => compare(left, right), + }, + ) + } + } +} + +#[cfg(test)] +mod tests; diff --git a/native/spark-expr/src/array_funcs/array_extrema/tests.rs b/native/spark-expr/src/array_funcs/array_extrema/tests.rs new file mode 100644 index 0000000000..4ba1d0f174 --- /dev/null +++ b/native/spark-expr/src/array_funcs/array_extrema/tests.rs @@ -0,0 +1,383 @@ +// 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. + +use super::SparkArrayExtrema; +use arrow::array::{ + Array, ArrayRef, Float64Array, Int32Array, ListArray, PrimitiveArray, StringArray, StructArray, +}; +use arrow::buffer::OffsetBuffer; +use arrow::datatypes::{Field, Float32Type, Float64Type, Int32Type}; +use datafusion::common::{config::ConfigOptions, ScalarValue}; +use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl}; +use std::sync::Arc; + +fn invoke(input: ColumnarValue, is_min: bool) -> ColumnarValue { + let udf = SparkArrayExtrema::new(is_min); + let input_type = input.data_type(); + let return_type = udf.return_type(std::slice::from_ref(&input_type)).unwrap(); + let number_rows = match &input { + ColumnarValue::Array(array) => array.len(), + ColumnarValue::Scalar(_) => 1, + }; + udf.invoke_with_args(ScalarFunctionArgs { + args: vec![input], + arg_fields: vec![Arc::new(Field::new("input", input_type, true))], + number_rows, + return_field: Arc::new(Field::new("result", return_type, true)), + config_options: Arc::new(ConfigOptions::default()), + }) + .unwrap() +} + +fn extrema(input: &dyn Array, is_min: bool) -> ArrayRef { + let ColumnarValue::Array(result) = + invoke(ColumnarValue::Array(input.slice(0, input.len())), is_min) + else { + panic!("array input must produce array output") + }; + result +} + +fn list(values: ArrayRef, offsets: &[i32]) -> ListArray { + ListArray::new( + Arc::new(Field::new_list_field(values.data_type().clone(), true)), + OffsetBuffer::new(offsets.to_vec().into()), + values, + None, + ) +} + +fn float64_bits(array: &dyn Array) -> Vec> { + array + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .map(|value| value.map(f64::to_bits)) + .collect() +} + +macro_rules! float_tests { + ($name:ident, $arrow_type:ty, $native:ident, $positive:expr, $negative:expr, $signaling:expr) => { + #[test] + fn $name() { + let positive = $native::from_bits($positive); + let negative = $native::from_bits($negative); + let signaling = $native::from_bits($signaling); + let mut long_nan = vec![Some(1.0); 67]; + long_nan[7] = Some(negative); + long_nan[8] = Some(positive); + let mut long_min_zero = vec![Some(1.0); 67]; + long_min_zero[7] = Some(-0.0); + long_min_zero[8] = Some(0.0); + let mut long_max_zero = vec![Some(-1.0); 67]; + long_max_zero[7] = Some(0.0); + long_max_zero[8] = Some(-0.0); + let cases = [ + (Some(vec![Some(0.0), Some(-0.0)]), Some(0.0), Some(0.0)), + ( + Some(vec![None, Some(-0.0), Some(0.0)]), + Some(-0.0), + Some(-0.0), + ), + ( + Some(vec![Some(positive), Some(negative)]), + Some(positive), + Some(positive), + ), + ( + Some(vec![Some(negative), Some(positive)]), + Some(negative), + Some(negative), + ), + ( + Some(vec![Some(signaling), Some(negative)]), + Some(signaling), + Some(signaling), + ), + ( + Some(vec![ + Some(negative), + Some($native::INFINITY), + Some($native::NEG_INFINITY), + ]), + Some($native::NEG_INFINITY), + Some(negative), + ), + ( + Some(vec![Some(3.0), None, Some(-2.0)]), + Some(-2.0), + Some(3.0), + ), + (None, None, None), + (Some(vec![]), None, None), + (Some(vec![None, None]), None, None), + // Exercise long, null-free lists: the old Arrow reduction changed winners. + (Some(long_nan), Some(1.0), Some(negative)), + (Some(long_min_zero), Some(-0.0), Some(1.0)), + (Some(long_max_zero), Some(-1.0), Some(0.0)), + ]; + for (row, min, max) in cases { + let input = ListArray::from_iter_primitive::<$arrow_type, _, _>([row]); + for (is_min, expected) in [(true, min), (false, max)] { + let result = extrema(&input, is_min); + let result = result + .as_any() + .downcast_ref::>() + .unwrap(); + assert_eq!( + result.iter().next().unwrap().map($native::to_bits), + expected.map($native::to_bits), + ); + } + } + } + }; +} + +float_tests!( + float32_preserves_first_winner, + Float32Type, + f32, + 0x7fc0_0001, + 0xffc0_0002, + 0x7f80_0001 +); +float_tests!( + float64_preserves_first_winner, + Float64Type, + f64, + 0x7ff8_0000_0000_0001, + 0xfff8_0000_0000_0002, + 0x7ff0_0000_0000_0001 +); + +#[test] +fn scalar_and_sliced_float_input() { + let input = ListArray::from_iter_primitive::([ + Some(vec![Some(99.0)]), + Some(vec![Some(-0.0), None, Some(0.0)]), + None, + ]) + .slice(1, 2); + for is_min in [true, false] { + assert_eq!( + float64_bits(extrema(&input, is_min).as_ref()), + vec![Some((-0.0f64).to_bits()), None] + ); + let scalar = ScalarValue::try_from_array(&input, 0).unwrap(); + let ColumnarValue::Scalar(result) = invoke(ColumnarValue::Scalar(scalar), is_min) else { + panic!("scalar input must produce scalar output") + }; + assert_eq!( + float64_bits(result.to_array().unwrap().as_ref()), + vec![Some((-0.0f64).to_bits())] + ); + } +} + +#[test] +fn nested_lists_use_spark_lexicographic_order_and_keep_original_bits() { + let positive = f64::from_bits(0x7ff8_0000_0000_0001); + let negative = f64::from_bits(0xfff8_0000_0000_0002); + let children = ListArray::from_iter_primitive::([ + Some(vec![Some(0.0), Some(positive)]), + Some(vec![Some(-0.0), Some(negative)]), + Some(vec![None]), + Some(vec![Some(f64::NEG_INFINITY)]), + Some(vec![]), + Some(vec![None]), + Some(vec![Some(-0.0)]), + Some(vec![Some(0.0), None]), + Some(vec![Some(negative)]), + Some(vec![Some(f64::INFINITY)]), + None, + Some(vec![Some(1.0)]), + None, + None, + ]); + let input = list(Arc::new(children.clone()), &[0, 2, 4, 6, 8, 10, 12, 14]); + for (is_min, winners) in [ + ( + true, + [Some(0), Some(2), Some(4), Some(6), Some(9), Some(11), None], + ), + ( + false, + [Some(0), Some(3), Some(5), Some(7), Some(8), Some(11), None], + ), + ] { + let result = extrema(&input, is_min); + let result = result.as_any().downcast_ref::().unwrap(); + for (row, winner) in winners.into_iter().enumerate() { + match winner { + Some(winner) => assert_eq!( + float64_bits(result.value(row).as_ref()), + float64_bits(children.value(winner).as_ref()), + ), + None => assert!(result.is_null(row)), + } + } + } +} + +#[test] +fn structs_compare_later_fields_without_normalizing_tied_floats() { + let nan = f64::from_bits(0xfff8_0000_0000_0002); + let columns: Vec = vec![ + Arc::new(Float64Array::from(vec![ + Some(-0.0), + Some(0.0), + Some(nan), + Some(f64::NAN), + None, + Some(f64::NEG_INFINITY), + ])), + Arc::new(Int32Array::from(vec![2, 1, 1, 1, 100, -100])), + ]; + let fields: Vec<_> = columns + .iter() + .enumerate() + .map(|(i, array)| { + Arc::new(Field::new( + format!("field_{i}"), + array.data_type().clone(), + true, + )) + }) + .collect(); + let children = StructArray::new(fields.into(), columns, None); + let input = list(Arc::new(children.clone()), &[0, 2, 4, 6]); + for (is_min, winners) in [(true, [1, 2, 4]), (false, [0, 2, 5])] { + let result = extrema(&input, is_min); + let result = result.as_any().downcast_ref::().unwrap(); + let floats = float64_bits(result.column(0).as_ref()); + let expected = float64_bits(children.column(0).as_ref()); + let ints = result + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + let expected_ints = children + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + for (row, winner) in winners.into_iter().enumerate() { + assert_eq!(floats[row], expected[winner]); + assert_eq!(ints.value(row), expected_ints.value(winner)); + } + } +} + +#[test] +fn nested_results_do_not_retain_losing_values() { + let rows = 512; + for (is_min, winner) in [(true, None), (true, Some(-0.0)), (false, Some(2.0))] { + let mut leaves = Vec::new(); + let mut offsets = vec![0]; + for _ in 0..rows { + leaves.extend(winner); + offsets.push(leaves.len() as i32); + leaves.extend(std::iter::repeat_n(1.0, 128)); + offsets.push(leaves.len() as i32); + } + let children = list(Arc::new(Float64Array::from(leaves)), &offsets); + let offsets: Vec = (0..=rows).map(|row| (row * 2) as i32).collect(); + let input = list(Arc::new(children), &offsets); + let result = extrema(&input, is_min); + let lists = result.as_any().downcast_ref::().unwrap(); + assert_eq!(lists.null_count(), 0); + for row in 0..rows { + assert_eq!( + float64_bits(lists.value(row).as_ref()), + winner + .into_iter() + .map(|v| Some(v.to_bits())) + .collect::>() + ); + } + assert!(result.get_buffer_memory_size() < 32 * 1024); + } +} + +#[test] +fn sparse_list_and_struct_results_bound_child_capacity() { + let rows = 8192; + let leaves = Arc::new(Float64Array::from(vec![-0.0; 1000])); + let child: ArrayRef = Arc::new(list(leaves, &[0, 1000])); + let fields = vec![Arc::new(Field::new( + "items", + child.data_type().clone(), + true, + ))]; + let structure: ArrayRef = Arc::new(StructArray::new( + fields.into(), + vec![Arc::clone(&child)], + None, + )); + let mut offsets = vec![1; rows + 1]; + offsets[0] = 0; + for child in [child, structure] { + let input = list(Arc::clone(&child), &offsets); + for is_min in [true, false] { + let result = extrema(&input, is_min); + assert_eq!(result.len(), rows); + assert_eq!(result.null_count(), rows - 1); + let selected = if let Some(structure) = result.as_any().downcast_ref::() { + structure.column(0).as_ref() + } else { + result.as_ref() + }; + let selected = selected + .as_any() + .downcast_ref::() + .unwrap() + .value(0); + assert_eq!( + float64_bits(selected.as_ref()), + vec![Some((-0.0f64).to_bits()); 1000] + ); + assert!(result.get_buffer_memory_size() < 256 * 1024); + } + } +} + +#[test] +fn delegates_non_floating_values() { + let input = ListArray::from_iter_primitive::([ + Some(vec![Some(3), None, Some(-2)]), + Some(vec![]), + ]); + for (is_min, expected) in [(true, vec![Some(-2), None]), (false, vec![Some(3), None])] { + let result = extrema(&input, is_min); + assert_eq!( + result.as_any().downcast_ref::().unwrap(), + &Int32Array::from(expected) + ); + } +} + +#[test] +fn empty_string_batch_retains_element_type() { + let input = list(Arc::new(StringArray::from(Vec::<&str>::new())), &[0]); + for is_min in [true, false] { + let result = extrema(&input, is_min); + assert!(result.is_empty()); + assert_eq!(result.data_type(), &input.value_type()); + } +} diff --git a/native/spark-expr/src/array_funcs/mod.rs b/native/spark-expr/src/array_funcs/mod.rs index b8877a93dc..b59dd44964 100644 --- a/native/spark-expr/src/array_funcs/mod.rs +++ b/native/spark-expr/src/array_funcs/mod.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +mod array_extrema; mod array_insert; mod array_position; mod array_slice; @@ -26,6 +27,7 @@ mod list_extract; mod sequence; mod size; +pub use array_extrema::SparkArrayExtrema; pub use array_insert::ArrayInsert; pub use array_position::SparkArrayPositionFunc; pub use array_slice::SparkArraySlice; diff --git a/native/spark-expr/src/comet_scalar_funcs.rs b/native/spark-expr/src/comet_scalar_funcs.rs index 8fe19f0aad..be7438d135 100644 --- a/native/spark-expr/src/comet_scalar_funcs.rs +++ b/native/spark-expr/src/comet_scalar_funcs.rs @@ -29,11 +29,11 @@ use crate::{ spark_ceil, spark_day_name, spark_decimal_div, spark_decimal_integral_div, spark_floor, spark_isnan, spark_lpad, spark_make_decimal, spark_month_name, spark_read_side_padding, spark_round, spark_rpad, spark_sequence, spark_to_time, spark_unhex, spark_unscaled_value, - EvalMode, SparkArrayPositionFunc, SparkArraySlice, SparkArraysOverlap, SparkContains, - SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc, SparkDayOfWeek, SparkFlatten, - SparkIcebergBucket, SparkIcebergTemporalTransform, SparkIcebergTruncate, SparkMakeDate, - SparkMakeInterval, SparkMakeTime, SparkMapExtract, SparkNextDay, SparkSecondsToTimestamp, - SparkSizeFunc, SparkWeekDay, + EvalMode, SparkArrayExtrema, SparkArrayPositionFunc, SparkArraySlice, SparkArraysOverlap, + SparkContains, SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc, SparkDayOfWeek, + SparkFlatten, SparkIcebergBucket, SparkIcebergTemporalTransform, SparkIcebergTruncate, + SparkMakeDate, SparkMakeInterval, SparkMakeTime, SparkMapExtract, SparkNextDay, + SparkSecondsToTimestamp, SparkSizeFunc, SparkWeekDay, }; use arrow::datatypes::DataType; use datafusion::common::{DataFusionError, Result as DataFusionResult}; @@ -311,6 +311,8 @@ pub fn create_comet_physical_fun_with_eval_mode( fn all_scalar_functions() -> Vec> { vec![ + Arc::new(ScalarUDF::new_from_impl(SparkArrayExtrema::new(true))), + Arc::new(ScalarUDF::new_from_impl(SparkArrayExtrema::new(false))), Arc::new(ScalarUDF::new_from_impl(SparkArrayPositionFunc::default())), Arc::new(ScalarUDF::new_from_impl(SparkArraySlice::default())), Arc::new(ScalarUDF::new_from_impl(SparkArraysOverlap::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 cca9f63f8b..8cdabebb85 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,30 @@ object CometArrayIntersect } } -object CometArrayMax extends CometExpressionSerde[ArrayMax] { +private object ArrayExtremaSupport extends CometTypeShim { + val incompatReason: String = + "Array extrema use binary string ordering for non-UTF8_BINARY collations " + + "(https://github.com/apache/datafusion-comet/issues/4496)." + + def getSupportLevel(elementType: DataType): SupportLevel = { + if (hasNonDefaultStringCollation(elementType)) { + // The dispatcher runs Spark's own comparison with the original collation IDs, including + // strings nested in arrays or structs. Keep the native bytewise comparison opt-in only. + Incompatible(Some(incompatReason)) + } else { + Compatible() + } + } +} + +object CometArrayMax extends CometExpressionSerde[ArrayMax] with CodegenDispatchFallback { + override def hasConditionalNativeDefault: Boolean = true + + override def getIncompatibleReasons(): Seq[String] = Seq(ArrayExtremaSupport.incompatReason) + + override def getSupportLevel(expr: ArrayMax): SupportLevel = + ArrayExtremaSupport.getSupportLevel(expr.dataType) + override def convert( expr: ArrayMax, inputs: Seq[Attribute], @@ -253,7 +276,14 @@ object CometArrayMax extends CometExpressionSerde[ArrayMax] { } } -object CometArrayMin extends CometExpressionSerde[ArrayMin] { +object CometArrayMin extends CometExpressionSerde[ArrayMin] with CodegenDispatchFallback { + override def hasConditionalNativeDefault: Boolean = true + + override def getIncompatibleReasons(): Seq[String] = Seq(ArrayExtremaSupport.incompatReason) + + override def getSupportLevel(expr: ArrayMin): SupportLevel = + ArrayExtremaSupport.getSupportLevel(expr.dataType) + override def convert( expr: ArrayMin, inputs: Seq[Attribute], diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_extrema_collation.sql b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_collation.sql new file mode 100644 index 0000000000..a4ce49df3d --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_collation.sql @@ -0,0 +1,57 @@ +-- 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. + +-- MinSparkVersion: 4.0 +-- Config: spark.comet.expression.ArrayMin.allowIncompatible=false +-- Config: spark.comet.expression.ArrayMax.allowIncompatible=false +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true + +statement +CREATE TABLE test_array_extrema_collation( + id int, a string, b string, x double, y double) USING parquet + +statement +INSERT INTO test_array_extrema_collation VALUES + (1, 'a', 'B', double('0.0'), double('-0.0')), + (2, 'B', 'a', double('-0.0'), double('0.0')), + (3, 'A', 'a', double('-0.0'), double('0.0')), + (4, NULL, 'B', NULL, double('0.0')), + (5, NULL, NULL, NULL, NULL) + +-- Binary strings and floating-point values remain native with the dispatcher enabled. +query expect_native(array_min,array_max) +SELECT id, array_min(array(a, b)), array_max(array(a, b)), + array_min(array(x, y)), array_max(array(x, y)), + array_min(array(named_struct('s', a, 'f', x), named_struct('s', b, 'f', y))), + array_max(array(named_struct('s', a, 'f', x), named_struct('s', b, 'f', y))) +FROM test_array_extrema_collation + +-- Case-insensitive ordering differs from binary ordering for 'a' and 'B'. +query expect_dispatch(array_min,array_max) +SELECT id, array_min(array(CAST(a AS STRING COLLATE UTF8_LCASE), CAST(b AS STRING COLLATE UTF8_LCASE))), + array_max(array(CAST(a AS STRING COLLATE UTF8_LCASE), CAST(b AS STRING COLLATE UTF8_LCASE))) +FROM test_array_extrema_collation + +-- Collation detection and Spark comparison also recurse through struct fields and arrays. +query expect_dispatch(array_min,array_max) +SELECT id, array_min(array( + named_struct('s', array(CAST(a AS STRING COLLATE UTF8_LCASE)), 'f', x), + named_struct('s', array(CAST(b AS STRING COLLATE UTF8_LCASE)), 'f', y))), + array_max(array( + named_struct('s', array(CAST(a AS STRING COLLATE UTF8_LCASE)), 'f', x), + named_struct('s', array(CAST(b AS STRING COLLATE UTF8_LCASE)), 'f', y))) +FROM test_array_extrema_collation diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_extrema_floating_point.sql b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_floating_point.sql new file mode 100644 index 0000000000..e0839b96f1 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_floating_point.sql @@ -0,0 +1,79 @@ +-- 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. + +-- Regression for https://github.com/apache/datafusion-comet/issues/5401. +-- Spark preserves the first equal extremum, including its zero sign. +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false +-- Config: spark.comet.expression.ArrayMin.allowIncompatible=false +-- Config: spark.comet.expression.ArrayMax.allowIncompatible=false +-- ConfigMatrix: spark.comet.exec.strictFloatingPoint=false,true + +statement +CREATE TABLE test_array_extrema_floating_point(id int, d array, f array) USING parquet + +statement +INSERT INTO test_array_extrema_floating_point VALUES + (1, array(double('0.0'), double('-0.0')), array(float('0.0'), float('-0.0'))), + (2, array(double('-0.0'), double('0.0')), array(float('-0.0'), float('0.0'))), + (3, array(NULL, double('-0.0'), double('0.0')), array(NULL, float('-0.0'), float('0.0'))), + (4, array(), array()), + (5, array(NULL, NULL), array(NULL, NULL)), + (6, NULL, NULL), + (7, array(double('NaN'), double('1.0')), array(float('NaN'), float('1.0'))), + (8, array(double('1.0'), double('NaN')), array(float('1.0'), float('NaN'))), + (9, array(double('NaN'), double('NaN')), array(float('NaN'), float('NaN'))), + (10, array(double('-Infinity'), double('Infinity')), array(float('-Infinity'), float('Infinity'))) + +query expect_native(array_min,array_max) +SELECT id, array_min(d), array_max(d), array_min(f), array_max(f) +FROM test_array_extrema_floating_point + +-- Constant folding is disabled by the harness, so literals exercise scalar evaluation. +query expect_native(array_min,array_max) +SELECT array_min(array(double('0.0'), double('-0.0'))), + array_max(array(double('-0.0'), double('0.0'))), + array_min(array(float('0.0'), float('-0.0'))), + array_max(array(float('-0.0'), float('0.0'))) + +-- Nested ties must compare later fields; fully equal elements retain their original bits. +statement +CREATE TABLE test_array_extrema_nested( + id int, d array>, s array>) USING parquet + +statement +INSERT INTO test_array_extrema_nested VALUES + (1, array(array(double('0.0'), double('1.0')), array(double('-0.0'), double('2.0'))), + array(named_struct('v', float('0.0'), 'payload', 1), named_struct('v', float('-0.0'), 'payload', 2))), + (2, array(array(double('-0.0')), array(double('0.0'))), + array(named_struct('v', float('-0.0'), 'payload', 1), named_struct('v', float('0.0'), 'payload', 1))), + (3, array(array(NULL), array(double('1.0'))), + array(named_struct('v', NULL, 'payload', 1), named_struct('v', float('1.0'), 'payload', 1))), + (4, array(array(double('NaN'), double('2.0')), array(double('NaN'), double('1.0'))), + array(named_struct('v', float('NaN'), 'payload', 2), named_struct('v', float('NaN'), 'payload', 1))), + (5, array(array(), array(NULL)), + array(NULL, named_struct('v', NULL, 'payload', NULL))) + +query expect_native(array_min,array_max) +SELECT id, array_min(d), array_max(d), array_min(s), array_max(s) +FROM test_array_extrema_nested + +-- Non-floating nested values retain nulls-first ordering. +query expect_native(array_min,array_max) +SELECT array_min(array(array(1, NULL), array(1, 0))), + array_max(array(array(1, NULL), array(1, 0))), + array_min(array(named_struct('k', 1, 'v', NULL), named_struct('k', 1, 'v', 'a'))), + array_max(array(named_struct('k', 1, 'v', NULL), named_struct('k', 1, 'v', 'a'))) diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_max.sql b/spark/src/test/resources/sql-tests/expressions/array/array_max.sql index e5f9db3e8a..77809447c1 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_max.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_max.sql @@ -15,13 +15,16 @@ -- specific language governing permissions and limitations -- under the License. +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false +-- Config: spark.comet.expression.ArrayMax.allowIncompatible=false + statement CREATE TABLE test_array_max(arr array) USING parquet statement INSERT INTO test_array_max VALUES (array(1, 2, 3)), (array(3, 1, 2)), (array()), (NULL), (array(NULL, 1, 2)), (array(-1, -2, -3)) -query spark_answer_only +query SELECT array_max(arr) FROM test_array_max -- literal arguments diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_min.sql b/spark/src/test/resources/sql-tests/expressions/array/array_min.sql index f3efb870ab..f1b217dada 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_min.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_min.sql @@ -15,13 +15,16 @@ -- specific language governing permissions and limitations -- under the License. +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false +-- Config: spark.comet.expression.ArrayMin.allowIncompatible=false + statement CREATE TABLE test_array_min(arr array) USING parquet statement INSERT INTO test_array_min VALUES (array(1, 2, 3)), (array(3, 1, 2)), (array()), (NULL), (array(NULL, 1, 2)), (array(-1, -2, -3)) -query spark_answer_only +query SELECT array_min(arr) FROM test_array_min -- literal arguments @@ -49,8 +52,8 @@ INSERT INTO test_array_min_double VALUES query SELECT array_min(arr) FROM test_array_min_double --- Spark treats +0.0 and -0.0 as equal and returns +0.0; Comet returns -0.0. --- Surfaced by https://github.com/apache/datafusion-comet/issues/5271 +-- Regression for https://github.com/apache/datafusion-comet/issues/5401: +-- Spark preserves the first equal zero (+0.0 here), and native execution must do the same. statement CREATE TABLE test_array_min_dbl_negzero(arr array) USING parquet @@ -58,7 +61,7 @@ statement INSERT INTO test_array_min_dbl_negzero VALUES (array(0.0, double('-0.0'), 1.0)) -query ignore(array_min signed-zero: Spark +0.0, Comet -0.0) +query SELECT array_min(arr) FROM test_array_min_dbl_negzero -- ===== FLOAT arrays with NaN/Infinity/-0.0 ===== @@ -79,8 +82,8 @@ INSERT INTO test_array_min_float VALUES query SELECT array_min(arr) FROM test_array_min_float --- Spark treats +0.0 and -0.0 as equal and returns +0.0; Comet returns -0.0. --- Surfaced by https://github.com/apache/datafusion-comet/issues/5271 +-- Regression for https://github.com/apache/datafusion-comet/issues/5401: +-- Spark preserves the first equal zero (+0.0 here), and native execution must do the same. statement CREATE TABLE test_array_min_flt_negzero(arr array) USING parquet @@ -88,5 +91,5 @@ statement INSERT INTO test_array_min_flt_negzero VALUES (array(CAST(0.0 AS FLOAT), float('-0.0'))) -query ignore(array_min signed-zero: Spark +0.0, Comet -0.0) +query SELECT array_min(arr) FROM test_array_min_flt_negzero diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index 8d0a04d7d6..1c4ad3ce15 100644 --- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala @@ -23,7 +23,7 @@ import scala.util.Random import org.apache.hadoop.fs.Path import org.apache.spark.sql.CometTestBase -import org.apache.spark.sql.catalyst.expressions.{ArrayAppend, ArrayExcept, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayRepeat} +import org.apache.spark.sql.catalyst.expressions.{ArrayAppend, ArrayExcept, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayRepeat} 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 @@ -631,6 +631,80 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp } } + test("array extrema - collations fall back when the dispatcher is disabled") { + assume(isSpark40Plus) + withParquetTable(Seq(("a", "B"), ("B", "a"), ("A", "a")), "collated_extrema") { + val a = "CAST(_1 AS STRING COLLATE UTF8_LCASE)" + val b = "CAST(_2 AS STRING COLLATE UTF8_LCASE)" + val inputs = Seq( + s"array($a, $b)", + s"array(named_struct('s', array($a)), named_struct('s', array($b)))") + withSQLConf( + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false", + CometConf.getExprAllowIncompatConfigKey(classOf[ArrayMin]) -> "false", + CometConf.getExprAllowIncompatConfigKey(classOf[ArrayMax]) -> "false") { + for (function <- Seq("array_min", "array_max"); input <- inputs) { + checkSparkAnswerAndFallbackReason( + s"SELECT $function($input) FROM collated_extrema", + "Array extrema use binary string ordering") + } + } + } + } + + test("array extrema - runtime NaN representations") { + withParquetTable(Seq((Float.NaN, Double.NaN)), "floating_point_extrema") { + for (strict <- Seq(false, true)) { + withSQLConf( + CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> strict.toString, + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false", + CometConf.getExprAllowIncompatConfigKey(classOf[ArrayMin]) -> "false", + CometConf.getExprAllowIncompatConfigKey(classOf[ArrayMax]) -> "false") { + for (function <- Seq("array_min", "array_max")) { + // Parquet canonicalizes NaNs. Negating the column after the scan supplies a + // different representation at runtime; ordinary SQL equality cannot check + // that extrema preserve the bits of the first equal NaN. + val query = sql(s""" + SELECT $function(array(-_1, _1)), $function(array(_1, -_1)), + $function(array(-_2, _2)), $function(array(_2, -_2)), + $function(array(-_1, CAST(1 AS FLOAT))), + $function(array(-_2, CAST(1 AS DOUBLE))), + $function(array(named_struct('v', -_1, 'n', 1), + named_struct('v', _1, 'n', 1))).v, + $function(array(named_struct('v', -_2, 'n', 1), + named_struct('v', _2, 'n', 1))).v + FROM floating_point_extrema + """) + checkSparkAnswerAndOperator(query) + val row = query.head() + val floatBits = java.lang.Float.floatToRawIntBits(Float.NaN) + val doubleBits = java.lang.Double.doubleToRawLongBits(Double.NaN) + val negativeFloatBits = floatBits | Int.MinValue + val negativeDoubleBits = doubleBits | Long.MinValue + assert(java.lang.Float.floatToRawIntBits(row.getFloat(0)) == negativeFloatBits) + assert(java.lang.Float.floatToRawIntBits(row.getFloat(1)) == floatBits) + assert(java.lang.Double.doubleToRawLongBits(row.getDouble(2)) == negativeDoubleBits) + assert(java.lang.Double.doubleToRawLongBits(row.getDouble(3)) == doubleBits) + val expectedFloatBits = if (function == "array_min") { + java.lang.Float.floatToRawIntBits(1.0f) + } else { + negativeFloatBits + } + val expectedDoubleBits = if (function == "array_min") { + java.lang.Double.doubleToRawLongBits(1.0d) + } else { + negativeDoubleBits + } + assert(java.lang.Float.floatToRawIntBits(row.getFloat(4)) == expectedFloatBits) + assert(java.lang.Double.doubleToRawLongBits(row.getDouble(5)) == expectedDoubleBits) + assert(java.lang.Float.floatToRawIntBits(row.getFloat(6)) == negativeFloatBits) + assert(java.lang.Double.doubleToRawLongBits(row.getDouble(7)) == negativeDoubleBits) + } + } + } + } + } + test("arrays_overlap - runtime NaN representations") { val floatNaN = java.lang.Float.intBitsToFloat(0x7fc01234 | Int.MinValue) val doubleNaN = java.lang.Double.longBitsToDouble(0x7ff8000000001234L | Long.MinValue)