diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index 31261b2542..08f5c0d1ac 100644 --- a/docs/source/contributor-guide/expression-audits/array_funcs.md +++ b/docs/source/contributor-guide/expression-audits/array_funcs.md @@ -90,19 +90,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, are flagged `Incompatible` ([#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 remain `Incompatible` ([#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 ffab055060..edee0a487d 100644 --- a/docs/source/user-guide/latest/compatibility/floating-point.md +++ b/docs/source/user-guide/latest/compatibility/floating-point.md @@ -27,3 +27,10 @@ So Comet adds additional normalization expression of NaN and zero for comparison to Spark in some cases, especially when the data contains both positive and negative zero. This is likely an edge 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. diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 192eae807e..c71eb96e8b 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 | 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_max` | ✅ | Native | Spark-compatible floating-point and nested ordering; non-default string collations fall back ([details](compatibility/expressions/array.md)) | +| `array_min` | ✅ | Native | Spark-compatible floating-point and nested ordering; non-default string collations fall back ([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 6faa9fec4e..beee7bde19 100644 --- a/native/spark-expr/Cargo.toml +++ b/native/spark-expr/Cargo.toml @@ -168,6 +168,10 @@ harness = false name = "arrays_overlap" harness = false +[[bench]] +name = "array_extrema" +harness = false + [[bench]] name = "checked_arithmetic" harness = false @@ -222,4 +226,4 @@ harness = false [[bench]] name = "cast_int_to_decimal" -harness = false \ No newline at end of file +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..0870b07713 --- /dev/null +++ b/native/spark-expr/benches/array_extrema.rs @@ -0,0 +1,187 @@ +// 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. + +//! Compares the Spark-compatible UDF with the DataFusion version pinned by Cargo.lock. +//! All timed inputs contain ordinary finite, nonzero values. Nested elements contain +//! no inner nulls: mixed zero signs, NaN payloads, and nested null ordering deliberately +//! differ from DataFusion and belong in the correctness tests, not parity benchmarks. +//! The null percentage controls both outer-row and immediate-child validity. +//! +//! Run the whole bounded matrix, or filter (for example) by float64 or nested_float64: +//! cargo bench -p datafusion-comet-spark-expr --bench array_extrema -- float64 + +use arrow::array::{ArrayRef, Float32Array, Float64Array, Int32Array, ListArray}; +use arrow::buffer::{NullBuffer, OffsetBuffer}; +use arrow::datatypes::Field; +use criterion::{ + criterion_group, criterion_main, measurement::WallTime, BenchmarkGroup, BenchmarkId, Criterion, + Throughput, +}; +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 valid(index: usize, null_percent: usize) -> bool { + (index * 17 + 23) % 100 >= null_percent +} + +fn list(values: ArrayRef, rows: usize, len: usize, null_percent: usize) -> ArrayRef { + let offsets: Vec = (0..=rows).map(|row| (row * len) as i32).collect(); + let nulls = (null_percent != 0).then(|| { + NullBuffer::from( + (0..rows) + .map(|row| valid(row, null_percent)) + .collect::>(), + ) + }); + Arc::new(ListArray::new( + Arc::new(Field::new_list_field(values.data_type().clone(), true)), + OffsetBuffer::new(offsets.into()), + values, + nulls, + )) +} + +fn finite_value(index: usize) -> i32 { + ((index * 104_729 + 51) % 1_000_003 + 1) as i32 +} + +fn primitive_input(kind: &str, rows: usize, len: usize, null_percent: usize) -> ArrayRef { + let values = (0..rows * len).map(|i| valid(i, null_percent).then(|| finite_value(i))); + let values: ArrayRef = match kind { + "float32" => Arc::new(Float32Array::from_iter( + values.map(|v| v.map(|v| v as f32 / 8.0)), + )), + "float64" => Arc::new(Float64Array::from_iter( + values.map(|v| v.map(|v| f64::from(v) / 8.0)), + )), + "int32_control" => Arc::new(Int32Array::from_iter(values)), + _ => unreachable!(), + }; + list(values, rows, len, null_percent) +} + +fn nested_input(rows: usize, len: usize, null_percent: usize) -> ArrayRef { + let children = rows * len; + let values: ArrayRef = Arc::new(Float64Array::from_iter_values( + (0..children * 4).map(|i| f64::from(finite_value(i)) / 8.0), + )); + // Null immediate children are skipped by both implementations; there are no + // null float values inside a valid child list, so recursive ordering agrees. + list( + list(values, children, 4, null_percent), + rows, + len, + null_percent, + ) +} + +fn args(input: &ArrayRef, udf: &ScalarUDF) -> ScalarFunctionArgs { + 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", + udf.return_type(&[input.data_type().clone()]).unwrap(), + true, + )), + config_options: Arc::new(ConfigOptions::default()), + } +} + +fn bench_case( + group: &mut BenchmarkGroup<'_, WallTime>, + input: ArrayRef, + len: usize, + null_percent: usize, +) { + let case = format!("rows={}_len={len}_null={null_percent}pct", input.len()); + group.throughput(Throughput::Elements((input.len() * len) as u64)); + for is_min in [true, false] { + let operation = if is_min { "min" } else { "max" }; + let comet = ScalarUDF::from(SparkArrayExtrema::new(is_min)); + let datafusion = if is_min { + array_min_udf() + } else { + array_max_udf() + }; + let args = args(&input, &comet); + // Validate every fixture before timing. This is a parity/control check for + // ordinary data only, not an oracle for Spark's special-value semantics. + let comet_result = comet + .invoke_with_args(args.clone()) + .unwrap() + .into_array(input.len()) + .unwrap(); + let datafusion_result = datafusion + .invoke_with_args(args.clone()) + .unwrap() + .into_array(input.len()) + .unwrap(); + assert_eq!( + comet_result.to_data(), + datafusion_result.to_data(), + "{operation}/{case}" + ); + for (name, udf) in [("comet", &comet), ("datafusion", datafusion.as_ref())] { + group.bench_function( + BenchmarkId::new(format!("{operation}_{name}"), &case), + |b| b.iter(|| black_box(udf.invoke_with_args(black_box(args.clone())).unwrap())), + ); + } + } +} + +fn criterion_benchmark(c: &mut Criterion) { + // 104 cases with one-second measurements and a short warmup: roughly two + // minutes on an idle machine. CLI filters can select individual dimensions. + for kind in ["float32", "float64", "int32_control", "nested_float64"] { + let mut group = c.benchmark_group(format!("array_extrema/{kind}")); + group.sample_size(20); + group.warm_up_time(Duration::from_millis(250)); + group.measurement_time(Duration::from_secs(1)); + let (lengths, null_percentages): (&[usize], &[usize]) = match kind { + "float32" | "float64" => (&[8, 32, 1024], &[0, 10, 50]), + "int32_control" => (&[8, 1024], &[0, 50]), + _ => (&[8, 64], &[0, 50]), + }; + for &len in lengths { + let rows = (65_536 / len).min(4096); + for &null_percent in null_percentages { + let input = if kind == "nested_float64" { + nested_input(rows, len, null_percent) + } else { + primitive_input(kind, rows, len, null_percent) + }; + bench_case(&mut group, input, len, null_percent); + } + } + 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..854cc337d7 --- /dev/null +++ b/native/spark-expr/src/array_funcs/array_extrema.rs @@ -0,0 +1,418 @@ +// 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, + FixedSizeListArray, GenericListArray, GenericListViewArray, MutableArrayData, OffsetSizeTrait, + PrimitiveArray, PrimitiveBuilder, StructArray, UInt64Array, +}; +use arrow::buffer::NullBuffer; +use arrow::compute::{cast, 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 !needs_spark_ordering(&element_type) { + 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()?, + }; + let result = match array.data_type() { + DataType::List(_) => array_extrema(array.as_list::(), self.is_min)?, + DataType::LargeList(_) => array_extrema(array.as_list::(), self.is_min)?, + other => return exec_err!("{} does not support type {other}", self.name()), + }; + + if is_scalar { + Ok(ColumnarValue::Scalar(ScalarValue::try_from_array( + &result, 0, + )?)) + } else { + Ok(ColumnarValue::Array(result)) + } + } +} + +fn needs_spark_ordering(data_type: &DataType) -> bool { + matches!( + data_type, + DataType::Float32 + | DataType::Float64 + | DataType::List(_) + | DataType::LargeList(_) + | DataType::FixedSizeList(_, _) + | DataType::ListView(_) + | DataType::LargeListView(_) + | DataType::Struct(_) + | DataType::Dictionary(_, _) + ) +} + +fn array_extrema( + array: &GenericListArray, + 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. +fn float_extrema( + array: &GenericListArray, + 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: &GenericListArray, + is_min: bool, +) -> Result { + let values = array.values(); + let compare = spark_comparator(values)?; + // Dictionary keys can refer to null values even when the keys themselves are valid. + let nulls = values.logical_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.as_ref().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 u64)); + } + // Take from the original values, not comparator-normalized or reconstructed values. + // This preserves nested fields, dictionary types, signed zeros, and NaN payloads. + take_extrema_values(values, &UInt64Array::from(indices)) +} + +fn take_extrema_values(values: &ArrayRef, indices: &UInt64Array) -> Result { + let nulls = || { + Some( + indices + .iter() + .map(|index| index.is_some_and(|index| values.is_valid(index as usize))) + .collect::(), + ) + }; + match values.data_type() { + DataType::List(field) | DataType::LargeList(field) => { + // Winners are distinct, so flat-list take cannot amplify the source here. + // Nested children can amplify capacity recursively even with one output row. + if indices.len() <= values.len() && !field.data_type().is_nested() { + let mut result = take(values.as_ref(), indices, None)?; + // Arrow estimates child capacity from all inputs, including large losers. + // Release unused capacity before downstream operators reserve this result. + result.shrink_to_fit(); + return Ok(result); + } + let data = values.to_data(); + // Start children empty, including fixed-width children nested inside lists. + 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.extend(0, index as usize, index as usize + 1), + None => result.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::>>()?; + Ok(Arc::new(StructArray::try_new_with_length( + fields.clone(), + columns, + nulls(), + indices.len(), + )?)) + } + DataType::FixedSizeList(field, size) => { + let child_indices: UInt64Array = indices + .iter() + .flat_map(|index| { + let index = index.filter(|&index| values.is_valid(index as usize)); + (0..*size as u64) + .map(move |offset| index.map(|index| index * *size as u64 + offset)) + }) + .collect(); + let children = + take_extrema_values(values.as_fixed_size_list().values(), &child_indices)?; + Ok(Arc::new(FixedSizeListArray::try_new( + Arc::clone(field), + *size, + children, + nulls(), + )?)) + } + // Preserve dictionary keys and shared ListView values with their existing kernels. + _ => 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(_) => list_comparator(array.as_list::()), + DataType::LargeList(_) => list_comparator(array.as_list::()), + DataType::ListView(_) => list_view_comparator(array.as_list_view::()), + DataType::LargeListView(_) => list_view_comparator(array.as_list_view::()), + DataType::FixedSizeList(_, _) => { + let array = array.as_fixed_size_list(); + let compare = spark_comparator(array.values())?; + let size = array.value_length() as usize; + Ok(nulls_first(array.logical_nulls(), move |left, right| { + compare_ranges(left * size, size, right * size, size, &compare) + })) + } + DataType::Struct(_) => { + let array = array.as_struct(); + let fields = array + .columns() + .iter() + .map(spark_comparator) + .collect::>>()?; + Ok(nulls_first(array.logical_nulls(), move |left, right| { + fields + .iter() + .map(|compare| compare(left, right)) + .find(|&ordering| ordering != Ordering::Equal) + .unwrap_or(Ordering::Equal) + })) + } + DataType::Dictionary(_, value_type) => { + // Decode only for comparisons. Recursing after decoding also covers + // dictionaries whose values are nested arrays or structs with floats. + spark_comparator(&cast(array.as_ref(), value_type)?) + } + _ => 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.logical_nulls(), 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 list_comparator(array: &GenericListArray) -> Result { + let compare = spark_comparator(array.values())?; + let offsets = array.offsets().clone(); + Ok(nulls_first(array.logical_nulls(), move |left, right| { + let left_start = offsets[left].as_usize(); + let right_start = offsets[right].as_usize(); + compare_ranges( + left_start, + offsets[left + 1].as_usize() - left_start, + right_start, + offsets[right + 1].as_usize() - right_start, + &compare, + ) + })) +} + +fn list_view_comparator( + array: &GenericListViewArray, +) -> Result { + let compare = spark_comparator(array.values())?; + let offsets = array.offsets().clone(); + let sizes = array.sizes().clone(); + Ok(nulls_first(array.logical_nulls(), move |left, right| { + compare_ranges( + offsets[left].as_usize(), + sizes[left].as_usize(), + offsets[right].as_usize(), + sizes[right].as_usize(), + &compare, + ) + })) +} + +fn compare_ranges( + left_start: usize, + left_len: usize, + right_start: usize, + right_len: usize, + compare: &DynComparator, +) -> Ordering { + 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) +} + +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..bd532774d8 --- /dev/null +++ b/native/spark-expr/src/array_funcs/array_extrema/tests.rs @@ -0,0 +1,1171 @@ +// 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, BinaryArray, DictionaryArray, FixedSizeListArray, Float32Array, Float64Array, + Int32Array, Int8Array, LargeListArray, LargeListViewArray, ListArray, ListViewArray, NullArray, + PrimitiveArray, StringArray, StructArray, +}; +use arrow::buffer::{NullBuffer, OffsetBuffer}; +use arrow::datatypes::{DataType, Field, Float32Type, Float64Type, Int32Type, Int8Type}; +use datafusion::common::{config::ConfigOptions, ScalarValue}; +use datafusion::logical_expr::{ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl}; +use std::cmp::Ordering; +use std::sync::Arc; + +fn invoke(input: ColumnarValue, is_min: bool, number_rows: usize) -> 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(); + 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 result = invoke( + ColumnarValue::Array(input.slice(0, input.len())), + is_min, + input.len(), + ); + let ColumnarValue::Array(result) = result else { + panic!("array input must produce array output") + }; + result +} + +fn list(values: ArrayRef, offsets: &[i32], validity: Option>) -> ListArray { + ListArray::new( + Arc::new(Field::new_list_field(values.data_type().clone(), true)), + OffsetBuffer::new(offsets.to_vec().into()), + values, + validity.map(NullBuffer::from), + ) +} + +fn large_list(input: &ListArray) -> LargeListArray { + LargeListArray::new( + Arc::new(Field::new_list_field(input.value_type(), true)), + OffsetBuffer::new(input.offsets().iter().map(|&x| i64::from(x)).collect()), + Arc::clone(input.values()), + input.nulls().cloned(), + ) +} + +/// Arrow's value equality alone does not establish signed-zero or NaN-payload preservation. +fn assert_same_value(actual: &dyn Array, ai: usize, expected: &dyn Array, ei: usize) { + assert_eq!(actual.data_type(), expected.data_type()); + let actual_null = actual.logical_nulls().is_some_and(|n| n.is_null(ai)); + let expected_null = expected.logical_nulls().is_some_and(|n| n.is_null(ei)); + assert_eq!(actual_null, expected_null); + if actual_null { + return; + } + match actual.data_type() { + DataType::Float32 => assert_eq!( + actual + .as_any() + .downcast_ref::() + .unwrap() + .value(ai) + .to_bits(), + expected + .as_any() + .downcast_ref::() + .unwrap() + .value(ei) + .to_bits(), + ), + DataType::Float64 => assert_eq!( + actual + .as_any() + .downcast_ref::() + .unwrap() + .value(ai) + .to_bits(), + expected + .as_any() + .downcast_ref::() + .unwrap() + .value(ei) + .to_bits(), + ), + DataType::List(_) + | DataType::LargeList(_) + | DataType::FixedSizeList(_, _) + | DataType::ListView(_) + | DataType::LargeListView(_) => { + let value = |array: &dyn Array, index| match array.data_type() { + DataType::List(_) => array + .as_any() + .downcast_ref::() + .unwrap() + .value(index), + DataType::LargeList(_) => array + .as_any() + .downcast_ref::() + .unwrap() + .value(index), + DataType::FixedSizeList(_, _) => array + .as_any() + .downcast_ref::() + .unwrap() + .value(index), + DataType::ListView(_) => array + .as_any() + .downcast_ref::() + .unwrap() + .value(index), + _ => array + .as_any() + .downcast_ref::() + .unwrap() + .value(index), + }; + let actual = value(actual, ai); + let expected = value(expected, ei); + assert_eq!(actual.len(), expected.len()); + for i in 0..actual.len() { + assert_same_value(actual.as_ref(), i, expected.as_ref(), i); + } + } + DataType::Struct(_) => { + let actual = actual.as_any().downcast_ref::().unwrap(); + let expected = expected.as_any().downcast_ref::().unwrap(); + for (actual, expected) in actual.columns().iter().zip(expected.columns()) { + assert_same_value(actual.as_ref(), ai, expected.as_ref(), ei); + } + } + DataType::Dictionary(_, _) => { + let actual = actual + .as_any() + .downcast_ref::>() + .unwrap(); + let expected = expected + .as_any() + .downcast_ref::>() + .unwrap(); + // Taking the original element should preserve both the dictionary key and its value. + assert_eq!(actual.keys().value(ai), expected.keys().value(ei)); + assert_same_value( + actual.values().as_ref(), + actual.key(ai).unwrap(), + expected.values().as_ref(), + expected.key(ei).unwrap(), + ); + } + _ => assert_eq!( + ScalarValue::try_from_array(actual, ai).unwrap(), + ScalarValue::try_from_array(expected, ei).unwrap(), + ), + } +} + +fn assert_winners( + input: &dyn Array, + children: &dyn Array, + minima: &[Option], + maxima: &[Option], +) { + for (is_min, expected) in [(true, minima), (false, maxima)] { + let result = extrema(input, is_min); + assert_eq!(result.len(), expected.len()); + assert_eq!(result.data_type(), children.data_type()); + for (row, expected) in expected.iter().enumerate() { + match expected { + Some(index) => assert_same_value(result.as_ref(), row, children, *index), + None => assert!(result.logical_nulls().unwrap().is_null(row)), + } + } + } +} + +#[test] +fn sparse_nested_results_bound_child_capacity() { + let sparse_rows = 8192; + let child_count = 1000; + let leaves: ArrayRef = Arc::new(Float64Array::from_iter_values((0..child_count).map( + |i| match i { + 0 => -0.0, + 1 => f64::from_bits(0xfff8_0000_0000_1234), + _ => i as f64, + }, + ))); + let inner = list(Arc::clone(&leaves), &[0, child_count], None); + let inner: ArrayRef = Arc::new(inner); + let dictionary: ArrayRef = Arc::new(DictionaryArray::::new( + Int8Array::from(vec![127]), + Arc::new(Float64Array::from_iter_values((0..128).map(f64::from))), + )); + let view: ArrayRef = Arc::new(ListViewArray::new( + Arc::new(Field::new_list_field(DataType::Float64, false)), + vec![0].into(), + vec![child_count].into(), + Arc::clone(&leaves), + None, + )); + let fixed: ArrayRef = Arc::new(FixedSizeListArray::new( + Arc::new(Field::new_list_field(DataType::Float64, false)), + child_count, + leaves, + None, + )); + let mut sparse_offsets = vec![1; sparse_rows + 1]; + sparse_offsets[0] = 0; + let sparse = list(Arc::clone(&fixed), &sparse_offsets, None); + let deep: ArrayRef = Arc::new(list(Arc::new(sparse), &[0, sparse_rows as i32], None)); + let children = [ + Arc::clone(&inner), + Arc::new(large_list( + inner.as_any().downcast_ref::().unwrap(), + )) as ArrayRef, + Arc::new(list(fixed, &[0, 1], None)) as ArrayRef, + Arc::new(FixedSizeListArray::new( + Arc::new(Field::new_list_field(inner.data_type().clone(), true)), + 1, + Arc::clone(&inner), + None, + )) as ArrayRef, + Arc::new(StructArray::new( + vec![ + Arc::new(Field::new("items", inner.data_type().clone(), true)), + Arc::new(Field::new( + "dictionary", + dictionary.data_type().clone(), + false, + )), + Arc::new(Field::new("view", view.data_type().clone(), true)), + ] + .into(), + vec![inner, dictionary, view], + None, + )) as ArrayRef, + Arc::new(FixedSizeListArray::new( + Arc::new(Field::new_list_field(DataType::Float64, false)), + 0, + Arc::new(Float64Array::from(Vec::::new())), + Some(NullBuffer::new_valid(1)), + )) as ArrayRef, + Arc::new(StructArray::new_empty_fields(1, None)) as ArrayRef, + deep, + ]; + for children in children { + for row_count in [1, sparse_rows] { + let mut offsets = vec![1; row_count + 1]; + offsets[0] = 0; + let input = list(Arc::clone(&children), &offsets, None); + for input in [Arc::new(large_list(&input)) as ArrayRef, Arc::new(input)] { + for is_min in [true, false] { + let result = extrema(input.as_ref(), is_min); + assert_eq!(result.len(), row_count); + assert_eq!(result.null_count(), row_count - 1); + assert_same_value(result.as_ref(), 0, children.as_ref(), 0); + assert!( + result.get_buffer_memory_size() < 256 * 1024, + "{} retained {} bytes for one selected nested value", + result.data_type(), + result.get_buffer_memory_size(), + ); + } + } + } + } +} + +#[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, None); + for children in [ + Arc::new(large_list(&children)) as ArrayRef, + Arc::new(children), + ] { + let offsets: Vec = (0..=rows).map(|row| (row * 2) as i32).collect(); + let input = list(Arc::clone(&children), &offsets, None); + for input in [Arc::new(large_list(&input)) as ArrayRef, Arc::new(input)] { + let result = extrema(input.as_ref(), is_min); + assert_eq!(result.len(), rows); + assert_eq!(result.null_count(), 0); + for row in 0..rows { + assert_same_value(result.as_ref(), row, children.as_ref(), row * 2); + } + assert!( + result.get_buffer_memory_size() < 32 * 1024, + "{} retained {} bytes for small nested winners", + result.data_type(), + result.get_buffer_memory_size(), + ); + } + } + } +} + +macro_rules! float_tests { + ($module:ident, $arrow_type:ty, $native:ident, $positive:expr, $negative:expr, $signaling:expr, $negative_signaling:expr) => { + mod $module { + use super::*; + + type Row = Option>>; + + fn nans() -> [$native; 4] { + [ + $native::from_bits($positive), + $native::from_bits($negative), + $native::from_bits($signaling), + $native::from_bits($negative_signaling), + ] + } + + // Independent stable-sort oracle, not the production scan or DataFusion extrema. + // Spark 3.5/4.0 SQLOrderingUtil compares all NaNs equal and greater than numbers; + // its x == y check makes the two zeros equal. ArrayMin/ArrayMax replace only on a + // strict comparison, so a stable sort in either direction must keep the first tie. + fn reference(row: &Row, is_min: bool) -> Option<$native> { + let mut values: Vec<_> = row.as_ref()?.iter().flatten().copied().collect(); + values.sort_by(|a, b| { + let order = match (a.is_nan(), b.is_nan()) { + (true, true) => Ordering::Equal, + (true, false) => Ordering::Greater, + (false, true) => Ordering::Less, + (false, false) => a.partial_cmp(b).unwrap(), + }; + if is_min { + order + } else { + order.reverse() + } + }); + values.first().copied() + } + + fn assert_bits(actual: &dyn Array, expected: &[Option<$native>]) { + let actual = actual + .as_any() + .downcast_ref::>() + .unwrap(); + let actual: Vec<_> = actual.iter().map(|x| x.map($native::to_bits)).collect(); + let expected: Vec<_> = expected.iter().map(|x| x.map($native::to_bits)).collect(); + assert_eq!(actual, expected); + } + + fn check_rows(rows: Vec) { + let input = ListArray::from_iter_primitive::<$arrow_type, _, _>(rows.clone()); + if rows.iter().flatten().flatten().all(Option::is_some) { + assert!(input.values().nulls().is_none()); + } + for is_min in [true, false] { + let expected: Vec<_> = rows.iter().map(|row| reference(row, is_min)).collect(); + assert_bits(extrema(&input, is_min).as_ref(), &expected); + assert_bits(extrema(&large_list(&input), is_min).as_ref(), &expected); + } + } + + #[test] + fn explicit_zero_nan_infinity_subnormal_and_null_cases() { + let [positive, negative, signaling, negative_signaling] = nans(); + let subnormal = $native::from_bits(1); + let cases: Vec<(Row, Option<$native>, Option<$native>)> = vec![ + (Some(vec![Some(0.0), Some(-0.0)]), Some(0.0), Some(0.0)), + (Some(vec![Some(-0.0), Some(0.0)]), Some(-0.0), Some(-0.0)), + ( + Some(vec![None, Some(0.0), None, Some(-0.0)]), + Some(0.0), + Some(0.0), + ), + ( + Some(vec![None, Some(-0.0), None, Some(0.0)]), + Some(-0.0), + Some(-0.0), + ), + ( + Some(vec![Some(positive), Some(negative), Some(signaling)]), + Some(positive), + Some(positive), + ), + ( + Some(vec![Some(negative), Some(signaling), Some(positive)]), + Some(negative), + Some(negative), + ), + ( + Some(vec![Some(signaling), Some(negative)]), + Some(signaling), + Some(signaling), + ), + ( + Some(vec![Some(negative_signaling), Some(positive)]), + Some(negative_signaling), + Some(negative_signaling), + ), + ( + Some(vec![ + Some(negative), + Some($native::INFINITY), + Some($native::NEG_INFINITY), + ]), + Some($native::NEG_INFINITY), + Some(negative), + ), + ( + Some(vec![ + None, + Some(signaling), + Some(-subnormal), + Some(-0.0), + Some(subnormal), + ]), + Some(-subnormal), + Some(signaling), + ), + ( + Some(vec![Some(subnormal), Some(0.0), Some(-0.0)]), + Some(0.0), + Some(subnormal), + ), + ( + Some(vec![Some(-subnormal), Some(-0.0), Some(0.0)]), + Some(-subnormal), + Some(-0.0), + ), + ( + Some(vec![Some(3.0), None, Some(-2.0), Some(3.0)]), + Some(-2.0), + Some(3.0), + ), + (None, None, None), + (Some(vec![]), None, None), + (Some(vec![None, None]), None, None), + (Some(vec![None, Some(-0.0)]), Some(-0.0), Some(-0.0)), + ]; + let rows: Vec<_> = cases.iter().map(|(row, _, _)| row.clone()).collect(); + let input = ListArray::from_iter_primitive::<$arrow_type, _, _>(rows.clone()); + for is_min in [true, false] { + let expected: Vec<_> = cases + .iter() + .map(|(_, min, max)| if is_min { *min } else { *max }) + .collect(); + assert_bits(extrema(&input, is_min).as_ref(), &expected); + } + check_rows(rows); + } + + #[test] + fn all_special_value_triples_match_stable_spark_ordering() { + let [positive, negative, signaling, negative_signaling] = nans(); + let values = [ + None, + Some(0.0), + Some(-0.0), + Some(positive), + Some(negative), + Some(signaling), + Some(negative_signaling), + Some($native::INFINITY), + Some($native::NEG_INFINITY), + Some($native::from_bits(1)), + Some(-$native::from_bits(1)), + Some(1.0), + Some(-1.0), + ]; + let mut rows = Vec::new(); + for a in values { + for b in values { + for c in values { + rows.push(Some(vec![a, b, c])); + } + } + } + // Repeat every non-null triple in long arrays as well, so future + // chunked reductions cannot change tie selection or special-value ordering. + let long_rows = rows + .iter() + .flatten() + .filter(|values| values.iter().all(Option::is_some)) + .map(|values| Some(values.iter().copied().cycle().take(67).collect())) + .collect(); + check_rows(long_rows); + check_rows(rows); + } + + #[test] + fn long_null_free_batches_handle_short_rows_and_slices() { + let [positive, negative, signaling, _] = nans(); + let mut long = vec![Some(1.0); 1024]; + long[7] = Some(-0.0); + long[8] = Some(0.0); + let rows = vec![ + Some(vec![Some(123.0); 128]), + None, + Some(vec![]), + Some(vec![Some(-0.0)]), + Some(vec![Some(-0.0), Some(0.0)]), + Some(vec![Some(negative), Some(signaling), Some(positive)]), + Some(long), + ]; + let input = ListArray::from_iter_primitive::<$arrow_type, _, _>(rows.clone()); + assert!(input.values().nulls().is_none()); + for is_min in [true, false] { + let expected: Vec<_> = rows.iter().map(|row| reference(row, is_min)).collect(); + assert_bits(extrema(&input, is_min).as_ref(), &expected); + assert_bits( + extrema(&input.slice(1, rows.len() - 1), is_min).as_ref(), + &expected[1..], + ); + assert_bits( + extrema(&large_list(&input).slice(1, rows.len() - 1), is_min).as_ref(), + &expected[1..], + ); + } + } + + #[test] + fn first_ties_survive_thresholds_and_lane_positions() { + let [positive, negative, signaling, _] = nans(); + for len in [31, 32, 33, 63, 64, 65, 66, 67, 257, 4096] { + let mut positions = vec![0, 7, 8, 15, 16, 31, 32, 63, 64, 65, 66, len - 1]; + positions.retain(|&p| p < len); + positions.sort_unstable(); + positions.dedup(); + let mut rows = vec![Some(vec![None; len])]; + for first_nan in [positive, negative, signaling] { + let mut row = vec![Some(positive); len]; + row[0] = Some(first_nan); + rows.push(Some(row)); + } + for &first in &positions { + for &second in positions.iter().filter(|&&p| p > first) { + for zero in [0.0, -0.0] { + for background in [1.0, -1.0] { + let mut row = vec![Some(background); len]; + row[first] = Some(zero); + row[second] = Some(-zero); + rows.push(Some(row.clone())); + for (i, value) in row.iter_mut().enumerate() { + if i % 3 == 0 && i != first && i != second { + *value = None; + } + } + rows.push(Some(row)); + } + } + let mut row = vec![Some(1.0); len]; + row[first] = Some(negative); + row[second] = Some(positive); + rows.push(Some(row)); + let mut row = vec![None; len]; + row[first] = Some(signaling); + row[second] = Some(negative); + rows.push(Some(row)); + } + } + // Also exercise child buffers without a null bitmap; mixing nullable + // and non-null rows in one batch does not establish that coverage. + let non_null_rows = rows + .iter() + .filter(|row| { + row.as_ref() + .is_some_and(|values| values.iter().all(Option::is_some)) + }) + .cloned() + .collect(); + check_rows(non_null_rows); + check_rows(rows); + } + } + + #[test] + fn sliced_children_outer_validity_and_hidden_null_children() { + let child = PrimitiveArray::<$arrow_type>::from_iter([ + Some(-999.0), + Some(999.0), + Some(0.0), + Some(-0.0), + Some($native::NEG_INFINITY), + Some(nans()[0]), + None, + None, + Some(-0.0), + Some(0.0), + Some(999.0), + ]) + .slice(1, 9); + let input = list( + Arc::new(child), + &[1, 3, 5, 5, 7, 9], + Some(vec![true, false, true, true, true]), + ); + for is_min in [true, false] { + assert_bits( + extrema(&input, is_min).as_ref(), + &[Some(0.0), None, None, None, Some(-0.0)], + ); + let sliced = input.slice(1, 4); + assert_bits( + extrema(&sliced, is_min).as_ref(), + &[None, None, None, Some(-0.0)], + ); + assert_bits( + extrema(&large_list(&input).slice(1, 4), is_min).as_ref(), + &[None, None, None, Some(-0.0)], + ); + let empty = input.slice(0, 0); + assert_bits(extrema(&empty, is_min).as_ref(), &[]); + assert_bits(extrema(&large_list(&empty), is_min).as_ref(), &[]); + } + } + + #[test] + fn scalar_input_returns_scalar_with_original_bits() { + let rows = vec![ + Some(vec![Some(-0.0), Some(0.0)]), + Some(vec![Some(nans()[2]), Some(nans()[1])]), + None, + Some(vec![]), + Some(vec![None]), + ]; + let input = ListArray::from_iter_primitive::<$arrow_type, _, _>(rows.clone()); + let large_input = large_list(&input); + for input in [&input as &dyn Array, &large_input] { + for (row, values) in rows.iter().enumerate() { + for is_min in [true, false] { + let scalar = ScalarValue::try_from_array(input, row).unwrap(); + let result = invoke(ColumnarValue::Scalar(scalar), is_min, 17); + let ColumnarValue::Scalar(result) = result else { + panic!("scalar input must produce scalar output") + }; + assert_bits( + result.to_array_of_size(1).unwrap().as_ref(), + &[reference(values, is_min)], + ); + } + } + } + } + + #[test] + fn dictionary_float_children_preserve_keys_bits_and_logical_nulls() { + let [positive, negative, signaling, _] = nans(); + let values: ArrayRef = Arc::new(PrimitiveArray::<$arrow_type>::from_iter([ + Some(0.0), + Some(-0.0), + Some(positive), + Some(negative), + Some(signaling), + None, + Some($native::NEG_INFINITY), + Some($native::INFINITY), + ])); + let keys = Int8Array::from(vec![ + Some(7), // Excluded by the child slice. + Some(0), + Some(1), + Some(1), + Some(0), + Some(2), + Some(3), + Some(4), + Some(5), + None, + Some(4), + Some(5), + None, + Some(3), + Some(7), + Some(6), + Some(6), + Some(7), + ]); + let dictionary = + DictionaryArray::::new(keys, Arc::clone(&values)).slice(1, 17); + let input = list( + Arc::new(dictionary), + &[0, 2, 4, 7, 10, 12, 15, 17, 17], + Some(vec![true, true, true, true, true, true, false, true]), + ); + assert_winners( + &input, + input.values().as_ref(), + &[ + Some(0), + Some(2), + Some(4), + Some(9), + None, + Some(14), + None, + None, + ], + &[ + Some(0), + Some(2), + Some(4), + Some(9), + None, + Some(12), + None, + None, + ], + ); + + // A valid dictionary key referring to a null value is null during nested + // comparison too: [null] sorts before [-infinity], not after it. + let dictionary = DictionaryArray::::new( + Int8Array::from(vec![5, 6, 0, 2, 1, 3]), + values, + ); + let inner = list(Arc::new(dictionary), &[0, 1, 2, 4, 6], None); + let outer = list(Arc::new(inner), &[0, 2, 4], None); + assert_winners( + &outer, + outer.values().as_ref(), + &[Some(0), Some(2)], + &[Some(1), Some(2)], + ); + } + } + }; +} + +float_tests!( + float32, + Float32Type, + f32, + 0x7fc0_0001, + 0xffc0_0002, + 0x7f80_0001, + 0xff80_0002 +); +float_tests!( + float64, + Float64Type, + f64, + 0x7ff8_0000_0000_0001, + 0xfff8_0000_0000_0002, + 0x7ff0_0000_0000_0001, + 0xfff0_0000_0000_0002 +); + +#[test] +fn nested_lists_keep_first_equal_bits_and_use_nulls_first() { + let positive = f64::from_bits(0x7ff8_0000_0000_0001); + let negative = f64::from_bits(0xfff8_0000_0000_0002); + let signaling = f64::from_bits(0x7ff0_0000_0000_0001); + let inner = ListArray::from_iter_primitive::([ + Some(vec![Some(0.0), Some(positive), None]), + Some(vec![Some(-0.0), Some(negative), None]), + Some(vec![Some(-0.0), Some(negative)]), + Some(vec![Some(0.0), Some(signaling)]), + Some(vec![None, Some(f64::INFINITY)]), + Some(vec![Some(f64::NEG_INFINITY), None]), + Some(vec![]), + Some(vec![None]), + None, + Some(vec![Some(-0.0)]), + Some(vec![Some(0.0)]), + Some(vec![Some(-0.0), None]), + Some(vec![Some(negative)]), + Some(vec![Some(f64::INFINITY)]), + None, + None, + ]); + let minima = [ + Some(0), + Some(2), + Some(4), + Some(6), + Some(9), + Some(10), + Some(13), + None, + ]; + let maxima = [ + Some(0), + Some(2), + Some(5), + Some(7), + Some(9), + Some(11), + Some(12), + None, + ]; + for inner in [ + Arc::new(inner.clone()) as ArrayRef, + Arc::new(large_list(&inner)), + ] { + let input = list(inner, &[0, 2, 4, 6, 8, 10, 12, 14, 16], None); + assert_winners(&input, input.values().as_ref(), &minima, &maxima); + assert_winners( + &large_list(&input), + input.values().as_ref(), + &minima, + &maxima, + ); + assert_winners(&input.slice(0, 0), input.values().as_ref(), &[], &[]); + } +} + +#[test] +fn nested_slices_and_null_parents_keep_original_children() { + let positive = f32::from_bits(0x7fc0_0001); + let negative = f32::from_bits(0xffc0_0002); + let inner = ListArray::from_iter_primitive::([ + Some(vec![Some(999.0)]), + Some(vec![Some(0.0), Some(positive)]), + Some(vec![Some(-0.0), Some(negative)]), + Some(vec![Some(f32::NEG_INFINITY)]), + Some(vec![Some(f32::INFINITY)]), + Some(vec![None]), + Some(vec![Some(-0.0)]), + ]) + .slice(1, 6); + let input = list( + Arc::new(inner), + &[0, 2, 4, 6], + Some(vec![true, false, true]), + ); + assert_winners( + &input, + input.values().as_ref(), + &[Some(0), None, Some(4)], + &[Some(0), None, Some(5)], + ); + let sliced = input.slice(1, 2); + assert_winners( + &sliced, + sliced.values().as_ref(), + &[None, Some(4)], + &[None, Some(5)], + ); +} + +#[test] +fn structs_compare_later_fields_and_preserve_original_nested_payloads() { + let positive32 = f32::from_bits(0x7fc0_0001); + let negative32 = f32::from_bits(0xffc0_0002); + let positive64 = f64::from_bits(0x7ff0_0000_0000_0001); + let negative64 = f64::from_bits(0xfff8_0000_0000_0002); + let floats: ArrayRef = Arc::new(Float32Array::from(vec![ + Some(0.0), + Some(-0.0), + Some(-0.0), + Some(0.0), + Some(0.0), + Some(-0.0), + Some(0.0), + Some(-0.0), + None, + Some(f32::NEG_INFINITY), + Some(positive32), + Some(negative32), + Some(f32::NEG_INFINITY), + Some(-0.0), + Some(0.0), + Some(-0.0), + ])); + let ints: ArrayRef = Arc::new(Int32Array::from(vec![ + 1, 1, 2, 1, 1, 1, 1, 1, 100, -100, 1, 1, 1, 1, 1, 1, + ])); + let strings: ArrayRef = Arc::new(StringArray::from(vec![ + "a", "a", "a", "a", "z", "a", "a", "a", "z", "a", "a", "a", "a", "a", "a", "a", + ])); + let binaries: ArrayRef = Arc::new(BinaryArray::from(vec![ + b"a".as_slice(), + b"a", + b"a", + b"a", + b"a", + b"z", + b"\x80", + b"\x7f", + b"a", + b"a", + b"a", + b"a", + b"a", + b"a", + b"a", + b"a", + ])); + let mut tails = vec![Some(vec![Some(0.0)]); 16]; + tails[0] = Some(vec![Some(positive64), Some(-0.0)]); + tails[1] = Some(vec![Some(negative64), Some(0.0)]); + tails[10] = Some(vec![None]); + tails[11] = Some(vec![Some(f64::NEG_INFINITY)]); + let tails: ArrayRef = Arc::new(ListArray::from_iter_primitive::(tails)); + let columns = vec![floats, ints, strings, binaries, tails]; + let fields: Vec<_> = columns + .iter() + .enumerate() + .map(|(i, array)| { + Arc::new(Field::new( + format!("field_{i}"), + array.data_type().clone(), + true, + )) + }) + .collect(); + let mut validity = vec![true; 16]; + for i in [12, 14, 15] { + validity[i] = false; + } + let structs = StructArray::new(fields.into(), columns, Some(NullBuffer::from(validity))); + let input = list(Arc::new(structs), &[0, 2, 4, 6, 8, 10, 12, 14, 16], None); + assert_winners( + &input, + input.values().as_ref(), + &[ + Some(0), + Some(3), + Some(5), + Some(7), + Some(8), + Some(10), + Some(13), + None, + ], + &[ + Some(0), + Some(2), + Some(4), + Some(6), + Some(9), + Some(11), + Some(13), + None, + ], + ); + assert_winners(&input.slice(0, 0), input.values().as_ref(), &[], &[]); + for is_min in [true, false] { + let scalar = ScalarValue::try_from_array(&input, 0).unwrap(); + let ColumnarValue::Scalar(result) = invoke(ColumnarValue::Scalar(scalar), is_min, 1) else { + panic!("nested scalar input must produce scalar output") + }; + assert_same_value( + result.to_array_of_size(1).unwrap().as_ref(), + 0, + input.values().as_ref(), + 0, + ); + } +} + +#[test] +fn non_floating_integer_and_nested_null_order_controls() { + let input = ListArray::from_iter_primitive::([ + Some(vec![Some(3), None, Some(-2), Some(3)]), + Some(vec![]), + Some(vec![None]), + None, + Some((0..64).map(|i| Some(i - 32)).collect()), + ]); + for (is_min, expected) in [ + (true, vec![Some(-2), None, None, None, Some(-32)]), + (false, vec![Some(3), None, None, None, Some(31)]), + ] { + let result = extrema(&input, is_min); + assert_eq!( + result.as_any().downcast_ref::().unwrap(), + &Int32Array::from(expected) + ); + } + let children = ListArray::from_iter_primitive::([ + Some(vec![None]), + Some(vec![Some(-1)]), + Some(vec![]), + Some(vec![None]), + ]); + let nested = list(Arc::new(children), &[0, 2, 4], None); + assert_winners( + &nested, + nested.values().as_ref(), + &[Some(0), Some(2)], + &[Some(1), Some(3)], + ); + let nulls = list(Arc::new(NullArray::new(3)), &[0, 3, 3], None); + for is_min in [true, false] { + let result = extrema(&nulls, is_min); + assert_eq!(result.data_type(), &DataType::Null); + assert_eq!(result.len(), 2); + // NullArray has no physical null bitmap; every element is logically null. + assert_eq!(result.logical_null_count(), 2); + } +} + +#[test] +fn empty_non_primitive_batches_retain_element_type() { + let children: [ArrayRef; 2] = [ + Arc::new(StringArray::from(Vec::<&str>::new())), + Arc::new(BinaryArray::from(Vec::<&[u8]>::new())), + ]; + for children in children { + let input = list(children, &[0], None); + assert_winners(&input, input.values().as_ref(), &[], &[]); + assert_winners(&large_list(&input), input.values().as_ref(), &[], &[]); + } +} + +#[test] +fn sliced_fixed_size_list_children_preserve_ties_and_null_order() { + let positive = f64::from_bits(0x7ff8_0000_0000_0001); + let negative = f64::from_bits(0xfff0_0000_0000_0002); + let field = Arc::new(Field::new_list_field(DataType::Float64, true)); + let values: ArrayRef = Arc::new(Float64Array::from(vec![ + Some(999.0), + Some(999.0), + Some(0.0), + Some(positive), + Some(-0.0), + Some(negative), + None, + Some(f64::INFINITY), + Some(f64::NEG_INFINITY), + None, + Some(0.0), + Some(0.0), + Some(-0.0), + Some(0.0), + ])); + let children = FixedSizeListArray::new( + Arc::clone(&field), + 2, + values, + Some(NullBuffer::from(vec![ + true, true, true, true, true, false, true, + ])), + ) + .slice(1, 6); + let input = list(Arc::new(children), &[0, 2, 4, 6], None); + assert_winners( + &input, + input.values().as_ref(), + &[Some(0), Some(2), Some(5)], + &[Some(0), Some(3), Some(5)], + ); + assert_winners( + &input.slice(1, 2), + input.values().as_ref(), + &[Some(2), Some(5)], + &[Some(3), Some(5)], + ); + + let empty_children = FixedSizeListArray::new( + field, + 0, + Arc::new(Float64Array::from(Vec::::new())), + Some(NullBuffer::from(vec![true, false, true])), + ); + let input = list(Arc::new(empty_children), &[0, 2, 3], None); + assert_winners( + &input, + input.values().as_ref(), + &[Some(0), Some(2)], + &[Some(0), Some(2)], + ); +} + +#[test] +fn sliced_list_view_children_support_nonmonotone_offsets() { + let positive = f64::from_bits(0x7ff8_0000_0000_0001); + let negative = f64::from_bits(0xfff0_0000_0000_0002); + let field = Arc::new(Field::new_list_field(DataType::Float64, true)); + let values: ArrayRef = Arc::new(Float64Array::from(vec![ + Some(0.0), + Some(positive), + Some(-0.0), + Some(negative), + None, + Some(f64::INFINITY), + Some(f64::NEG_INFINITY), + None, + ])); + let validity = Some(NullBuffer::from(vec![ + true, true, true, true, true, true, true, false, true, + ])); + let children = ListViewArray::new( + Arc::clone(&field), + vec![8, 2, 0, 4, 6, 0, 0, 0, 0].into(), + vec![0, 2, 2, 2, 2, 0, 1, 2, 2].into(), + Arc::clone(&values), + validity.clone(), + ) + .slice(1, 8); + let large_children = LargeListViewArray::new( + field, + vec![8, 2, 0, 4, 6, 0, 0, 0, 0].into(), + vec![0, 2, 2, 2, 2, 0, 1, 2, 2].into(), + values, + validity, + ) + .slice(1, 8); + for children in [Arc::new(children) as ArrayRef, Arc::new(large_children)] { + let input = list(children, &[0, 2, 4, 6, 8], None); + assert_winners( + &input, + input.values().as_ref(), + &[Some(0), Some(2), Some(4), Some(7)], + &[Some(0), Some(3), Some(5), Some(7)], + ); + assert_winners( + &large_list(&input).slice(1, 3), + input.values().as_ref(), + &[Some(2), Some(4), Some(7)], + &[Some(3), Some(5), Some(7)], + ); + } +} + +#[test] +fn return_field_is_nullable_and_retains_nested_field_metadata() { + let field = Field::new("value", DataType::Float64, false) + .with_metadata([(String::from("source"), String::from("extrema-test"))].into()); + let element_type = DataType::Struct(vec![field].into()); + let element_field = Arc::new(Field::new_list_field(element_type.clone(), false)); + for input_type in [ + DataType::List(Arc::clone(&element_field)), + DataType::LargeList(element_field), + ] { + let args = [Arc::new(Field::new("input", input_type, false))]; + for is_min in [true, false] { + let udf = SparkArrayExtrema::new(is_min); + assert_eq!(udf.name(), if is_min { "array_min" } else { "array_max" }); + let result = udf + .return_field_from_args(ReturnFieldArgs { + arg_fields: &args, + scalar_arguments: &[None], + }) + .unwrap(); + // Even a non-null input can be empty, so the result must remain nullable. + assert!(result.is_nullable()); + assert_eq!(result.data_type(), &element_type); + } + } +} diff --git a/native/spark-expr/src/array_funcs/mod.rs b/native/spark-expr/src/array_funcs/mod.rs index 0c2c68dc6d..25b1f5fcb6 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; @@ -25,6 +26,7 @@ mod get_array_struct_fields; mod list_extract; 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 dcb6b1906c..90fdb26ad4 100644 --- a/native/spark-expr/src/comet_scalar_funcs.rs +++ b/native/spark-expr/src/comet_scalar_funcs.rs @@ -27,9 +27,9 @@ 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_to_time, spark_unhex, spark_unscaled_value, EvalMode, - SparkArrayPositionFunc, SparkArraySlice, SparkArraysOverlap, SparkContains, SparkDateDiff, - SparkDateFromUnixDate, SparkDateTrunc, SparkFlatten, SparkMakeDate, SparkMakeInterval, - SparkMakeTime, SparkNextDay, SparkSecondsToTimestamp, SparkSizeFunc, + SparkArrayExtrema, SparkArrayPositionFunc, SparkArraySlice, SparkArraysOverlap, SparkContains, + SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc, SparkFlatten, SparkMakeDate, + SparkMakeInterval, SparkMakeTime, SparkNextDay, SparkSecondsToTimestamp, SparkSizeFunc, }; use arrow::datatypes::DataType; use datafusion::common::{DataFusionError, Result as DataFusionResult}; @@ -281,6 +281,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 748b1cee23..cd9414a32b 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -215,7 +215,26 @@ object CometArrayIntersect } } +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)) { + Incompatible(Some(incompatReason)) + } else { + Compatible() + } + } +} + object CometArrayMax extends CometExpressionSerde[ArrayMax] { + 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], @@ -229,6 +248,11 @@ object CometArrayMax extends CometExpressionSerde[ArrayMax] { } object CometArrayMin extends CometExpressionSerde[ArrayMin] { + 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_max.sql b/spark/src/test/resources/sql-tests/expressions/array/array_max.sql index e5f9db3e8a..9fd4306560 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 @@ -68,3 +71,44 @@ INSERT INTO test_array_max_float VALUES query SELECT array_max(arr) FROM test_array_max_float + +-- 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_max_negzero(d array, f array) USING parquet + +statement +INSERT INTO test_array_max_negzero VALUES + (array(double('-0.0'), double('0.0')), array(float('-0.0'), float('0.0'))) + +query +SELECT array_max(d), array_max(f) FROM test_array_max_negzero + +-- Default-mode non-floating native controls, including nested nulls-first ordering. +statement +CREATE TABLE test_array_max_nested_non_fp( + id int, a array>, s array>) USING parquet + +statement +INSERT INTO test_array_max_nested_non_fp VALUES + (1, array(array(1, NULL), array(1, 0)), + array(named_struct('k', 1, 'v', NULL), named_struct('k', 1, 'v', 'a'))), + (2, array(array(1, 0), array(1, NULL)), + array(named_struct('k', 1, 'v', 'a'), named_struct('k', 1, 'v', NULL))), + (3, array(array(), array(NULL)), + array(NULL, named_struct('k', NULL, 'v', 'a'))), + (4, array(array(1), array(1, NULL)), + array(named_struct('k', NULL, 'v', 'b'), named_struct('k', NULL, 'v', 'a'))), + (5, array(NULL, array(0)), array(NULL, named_struct('k', NULL, 'v', NULL))), + (6, array(NULL, NULL), array(NULL, NULL)), + (7, array(), array()), + (8, NULL, NULL) + +query +SELECT id, array_max(a), array_max(s) FROM test_array_max_nested_non_fp + +query +SELECT array_max(array(false, true, NULL)), + array_max(array('z', 'A', 'a')), + array_max(array(CAST(1.25 AS decimal(8, 2)), CAST(-2.5 AS decimal(8, 2)))), + array_max(array(DATE '2024-01-01', DATE '1969-12-31')) diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_max_collation.sql b/spark/src/test/resources/sql-tests/expressions/array/array_max_collation.sql new file mode 100644 index 0000000000..8451138c57 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_max_collation.sql @@ -0,0 +1,72 @@ +-- 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 +-- Non-binary collations still require Spark's ordering, including when nested. +-- Neither strict floating-point mode nor enabling the dispatcher may bypass this guard. +-- Config: spark.comet.expression.ArrayMax.allowIncompatible=false +-- ConfigMatrix: spark.comet.exec.strictFloatingPoint=false,true +-- ConfigMatrix: spark.comet.exec.scalaUDF.codegen.enabled=false,true + +statement +CREATE TABLE test_array_max_collation( + id int, a string, b string, x double, y double) USING parquet + +statement +INSERT INTO test_array_max_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, 'a', NULL, double('-0.0'), NULL), + (6, NULL, NULL, NULL, NULL) + +-- Binary string ordering remains native, including inside arrays and structs. +query +SELECT id, array_max(array(a, b)) FROM test_array_max_collation + +query +SELECT id, array_max(array(array(a), array(b))), + array_max(array(named_struct('s', a, 'f', x), named_struct('s', b, 'f', y))) +FROM test_array_max_collation + +-- Lowercase ordering differs from binary ordering for the column values 'a' and 'B'. +query expect_fallback(Array extrema use binary string ordering) +SELECT id, array_max(array( + CAST(a AS STRING COLLATE UTF8_LCASE), + CAST(b AS STRING COLLATE UTF8_LCASE))) +FROM test_array_max_collation + +query expect_fallback(Array extrema use binary string ordering) +SELECT id, array_max(array( + array(CAST(a AS STRING COLLATE UTF8_LCASE)), + array(CAST(b AS STRING COLLATE UTF8_LCASE)))) +FROM test_array_max_collation + +-- A floating-point field does not make a collated struct eligible for native ordering. +query expect_fallback(Array extrema use binary string ordering) +SELECT id, array_max(array( + named_struct('s', CAST(a AS STRING COLLATE UTF8_LCASE), 'f', x), + named_struct('s', CAST(b AS STRING COLLATE UTF8_LCASE), 'f', y))) +FROM test_array_max_collation + +-- The collation check must recurse through both struct fields and nested array elements. +query expect_fallback(Array extrema use binary string ordering) +SELECT id, 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_max_collation diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_max_floating_point.sql b/spark/src/test/resources/sql-tests/expressions/array/array_max_floating_point.sql new file mode 100644 index 0000000000..de6e7d34ef --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_max_floating_point.sql @@ -0,0 +1,189 @@ +-- 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 retains the first equal extremum: zero signs compare equal, as do NaNs. +-- Require exact native results in both floating-point modes without the codegen dispatcher. +-- The dictionary matrix varies the writer setting, not a guarantee of dictionary pages. +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false +-- Config: spark.comet.expression.ArrayMax.allowIncompatible=false +-- ConfigMatrix: spark.comet.exec.strictFloatingPoint=false,true +-- ConfigMatrix: parquet.enable.dictionary=false,true + +statement +CREATE TABLE test_array_max_floating_point(id int, d array, f array) USING parquet + +statement +INSERT INTO test_array_max_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'), NULL), array(NULL, float('0.0'), float('-0.0'), NULL)), + (4, array(NULL, double('-0.0'), double('0.0'), NULL), array(NULL, float('-0.0'), float('0.0'), NULL)), + (5, array(double('-0.0')), array(float('-0.0'))), + (6, array(double('0.0')), array(float('0.0'))), + (7, array(), array()), + (8, array(NULL), array(NULL)), + (9, array(NULL, NULL), array(NULL, NULL)), + (10, NULL, NULL), + (11, array(double('-3.0'), double('1.0'), double('2.0')), array(float('-3.0'), float('1.0'), float('2.0'))), + (12, array(NULL, double('-2.0'), double('4.0'), NULL), array(NULL, float('-2.0'), float('4.0'), NULL)), + (13, array(double('NaN'), double('1.0'), double('2.0')), array(float('NaN'), float('1.0'), float('2.0'))), + (14, array(double('1.0'), double('NaN'), double('2.0')), array(float('1.0'), float('NaN'), float('2.0'))), + (15, array(double('1.0'), double('2.0'), double('NaN')), array(float('1.0'), float('2.0'), float('NaN'))), + (16, array(double('NaN'), double('NaN')), array(float('NaN'), float('NaN'))), + (17, array(NULL, double('NaN'), NULL), array(NULL, float('NaN'), NULL)), + (18, array(double('-Infinity'), double('Infinity'), double('NaN')), array(float('-Infinity'), float('Infinity'), float('NaN'))), + (19, array(double('NaN'), double('Infinity'), double('-Infinity')), array(float('NaN'), float('Infinity'), float('-Infinity'))), + (20, array(double('-Infinity'), double('1.0'), double('Infinity')), array(float('-Infinity'), float('1.0'), float('Infinity'))), + (21, array(double('Infinity'), double('1.0'), double('-Infinity')), array(float('Infinity'), float('1.0'), float('-Infinity'))), + (22, array(double('Infinity'), double('Infinity')), array(float('Infinity'), float('Infinity'))), + (23, array(double('-Infinity'), double('-Infinity')), array(float('-Infinity'), float('-Infinity'))) + +query +SELECT id, array_max(d), array_max(f) FROM test_array_max_floating_point + +-- The harness disables constant folding, so literal inputs exercise native evaluation too. +query +SELECT array_max(array(double('0.0'), double('-0.0'))), + array_max(array(double('-0.0'), double('0.0'))), + array_max(array(float('0.0'), float('-0.0'))), + array_max(array(float('-0.0'), float('0.0'))) + +query +SELECT array_max(CAST(NULL AS array)), + array_max(CAST(NULL AS array)), + array_max(CAST(array() AS array)), + array_max(CAST(array() AS array)), + array_max(array(CAST(NULL AS double), CAST(NULL AS double))), + array_max(array(CAST(NULL AS float), CAST(NULL AS float))) + +query +SELECT array_max(array(double('NaN'), double('Infinity'), double('-Infinity'))), + array_max(array(float('NaN'), float('Infinity'), float('-Infinity'))), + array_max(array(double('NaN'), double('NaN'))), + array_max(array(float('NaN'), float('NaN'))), + array_max(array(double('Infinity'), double('-Infinity'))), + array_max(array(float('Infinity'), float('-Infinity'))) + +-- Exercise both tie orders around 32-element boundaries and across longer arrays. +-- Build the input during INSERT so the tested projection has only column arguments. +statement +CREATE TABLE test_array_max_floating_long( + n int, negative_first boolean, d array, f array, + nullable_d array, nullable_f array) USING parquet + +statement +INSERT INTO test_array_max_floating_long +SELECT n, negative_first, + concat(array(d0), array_repeat(d1, n - 1)), + concat(array(f0), array_repeat(f1, n - 1)), + concat(array_repeat(CAST(NULL AS double), n - 2), array(d0, d1)), + concat(array_repeat(CAST(NULL AS float), n - 2), array(f0, f1)) +FROM VALUES (31), (32), (33), (65), (129) AS sizes(n) +CROSS JOIN VALUES + (false, double('0.0'), double('-0.0'), float('0.0'), float('-0.0')), + (true, double('-0.0'), double('0.0'), float('-0.0'), float('0.0')) +AS signs(negative_first, d0, d1, f0, f1) + +query +SELECT n, negative_first, array_max(d), array_max(f), + array_max(nullable_d), array_max(nullable_f) +FROM test_array_max_floating_long + +-- Outer null elements are skipped; nulls inside an array sort before non-null elements. +-- Equal nested zeros and NaNs must allow comparison to continue to later elements. +statement +CREATE TABLE test_array_max_floating_nested( + id int, d array>, f array>) USING parquet + +statement +INSERT INTO test_array_max_floating_nested VALUES + (1, array(array(double('0.0')), array(double('-0.0'))), + array(array(float('0.0')), array(float('-0.0')))), + (2, array(array(double('-0.0')), array(double('0.0'))), + array(array(float('-0.0')), array(float('0.0')))), + (3, array(array(NULL, double('0.0')), array(NULL, double('-0.0'))), + array(array(NULL, float('0.0')), array(NULL, float('-0.0')))), + (4, array(array(NULL, double('-0.0')), array(NULL, double('0.0'))), + array(array(NULL, float('-0.0')), array(NULL, float('0.0')))), + (5, array(array(NULL), array(double('-Infinity'))), + array(array(NULL), array(float('-Infinity')))), + (6, array(array(double('-Infinity')), array(NULL)), + array(array(float('-Infinity')), array(NULL))), + (7, array(array(double('1.0'), NULL), array(double('1.0'), double('2.0'))), + array(array(float('1.0'), NULL), array(float('1.0'), float('2.0')))), + (8, array(array(double('1.0')), array(double('1.0'), NULL)), + array(array(float('1.0')), array(float('1.0'), NULL))), + (9, array(array(), array(NULL)), array(array(), array(NULL))), + (10, array(NULL, array(double('-0.0')), array(double('0.0'))), + array(NULL, array(float('-0.0')), array(float('0.0')))), + (11, array(NULL, NULL), array(NULL, NULL)), + (12, array(), array()), + (13, NULL, NULL), + (14, array(array(double('NaN'), double('2.0')), array(double('NaN'), double('1.0'))), + array(array(float('NaN'), float('2.0')), array(float('NaN'), float('1.0')))), + (15, array(array(double('0.0'), double('1.0')), array(double('-0.0'), double('2.0'))), + array(array(float('0.0'), float('1.0')), array(float('-0.0'), float('2.0')))) + +query +SELECT id, array_max(d), array_max(f) FROM test_array_max_floating_nested + +-- A zero tie in the first struct field must not override the ordering of its payload. +-- When every field compares equal, preserve the first struct, including its zero sign. +statement +CREATE TABLE test_array_max_floating_struct( + id int, d array>, f array>) USING parquet + +statement +INSERT INTO test_array_max_floating_struct VALUES + (1, array(named_struct('v', double('0.0'), 'payload', 1), named_struct('v', double('-0.0'), 'payload', 2)), + array(named_struct('v', float('0.0'), 'payload', 1), named_struct('v', float('-0.0'), 'payload', 2))), + (2, array(named_struct('v', double('-0.0'), 'payload', 2), named_struct('v', double('0.0'), 'payload', 1)), + array(named_struct('v', float('-0.0'), 'payload', 2), named_struct('v', float('0.0'), 'payload', 1))), + (3, array(named_struct('v', double('0.0'), 'payload', 1), named_struct('v', double('-0.0'), 'payload', 1)), + array(named_struct('v', float('0.0'), 'payload', 1), named_struct('v', float('-0.0'), 'payload', 1))), + (4, array(named_struct('v', double('-0.0'), 'payload', 1), named_struct('v', double('0.0'), 'payload', 1)), + array(named_struct('v', float('-0.0'), 'payload', 1), named_struct('v', float('0.0'), 'payload', 1))), + (5, array(named_struct('v', NULL, 'payload', 2), named_struct('v', double('-Infinity'), 'payload', 1)), + array(named_struct('v', NULL, 'payload', 2), named_struct('v', float('-Infinity'), 'payload', 1))), + (6, array(named_struct('v', double('0.0'), 'payload', NULL), named_struct('v', double('-0.0'), 'payload', 1)), + array(named_struct('v', float('0.0'), 'payload', NULL), named_struct('v', float('-0.0'), 'payload', 1))), + (7, array(named_struct('v', double('NaN'), 'payload', 2), named_struct('v', double('NaN'), 'payload', 1)), + array(named_struct('v', float('NaN'), 'payload', 2), named_struct('v', float('NaN'), 'payload', 1))), + (8, array(NULL, named_struct('v', NULL, 'payload', NULL)), + array(NULL, named_struct('v', NULL, 'payload', NULL))), + (9, array(NULL, NULL), array(NULL, NULL)), + (10, array(), array()), + (11, NULL, NULL) + +query +SELECT id, array_max(d), array_max(f) FROM test_array_max_floating_struct + +-- Recurse through a list of structs, not only structs or lists of primitive elements. +query +SELECT id, array_max(array(array(d[0]), array(d[1]))), + array_max(array(array(f[0]), array(f[1]))) +FROM test_array_max_floating_struct WHERE id IN (1, 2, 3, 4) + +-- Recurse in the other direction too: an array-valued struct field tied on zero. +query +SELECT array_max(array( + named_struct('v', array(double('0.0')), 'payload', 1), + named_struct('v', array(double('-0.0')), 'payload', 2))), + array_max(array( + named_struct('v', array(float('0.0')), 'payload', 1), + named_struct('v', array(float('-0.0')), 'payload', 2))) 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..b2d1f44e41 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,34 @@ 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 + +-- Default-mode non-floating native controls, including nested nulls-first ordering. +statement +CREATE TABLE test_array_min_nested_non_fp( + id int, a array>, s array>) USING parquet + +statement +INSERT INTO test_array_min_nested_non_fp VALUES + (1, array(array(1, NULL), array(1, 0)), + array(named_struct('k', 1, 'v', NULL), named_struct('k', 1, 'v', 'a'))), + (2, array(array(1, 0), array(1, NULL)), + array(named_struct('k', 1, 'v', 'a'), named_struct('k', 1, 'v', NULL))), + (3, array(array(), array(NULL)), + array(NULL, named_struct('k', NULL, 'v', 'a'))), + (4, array(array(1), array(1, NULL)), + array(named_struct('k', NULL, 'v', 'b'), named_struct('k', NULL, 'v', 'a'))), + (5, array(NULL, array(0)), array(NULL, named_struct('k', NULL, 'v', NULL))), + (6, array(NULL, NULL), array(NULL, NULL)), + (7, array(), array()), + (8, NULL, NULL) + +query +SELECT id, array_min(a), array_min(s) FROM test_array_min_nested_non_fp + +query +SELECT array_min(array(false, true, NULL)), + array_min(array('z', 'A', 'a')), + array_min(array(CAST(1.25 AS decimal(8, 2)), CAST(-2.5 AS decimal(8, 2)))), + array_min(array(DATE '2024-01-01', DATE '1969-12-31')) diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_min_collation.sql b/spark/src/test/resources/sql-tests/expressions/array/array_min_collation.sql new file mode 100644 index 0000000000..42c7678e28 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_min_collation.sql @@ -0,0 +1,72 @@ +-- 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 +-- Non-binary collations still require Spark's ordering, including when nested. +-- Neither strict floating-point mode nor enabling the dispatcher may bypass this guard. +-- Config: spark.comet.expression.ArrayMin.allowIncompatible=false +-- ConfigMatrix: spark.comet.exec.strictFloatingPoint=false,true +-- ConfigMatrix: spark.comet.exec.scalaUDF.codegen.enabled=false,true + +statement +CREATE TABLE test_array_min_collation( + id int, a string, b string, x double, y double) USING parquet + +statement +INSERT INTO test_array_min_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, 'a', NULL, double('-0.0'), NULL), + (6, NULL, NULL, NULL, NULL) + +-- Binary string ordering remains native, including inside arrays and structs. +query +SELECT id, array_min(array(a, b)) FROM test_array_min_collation + +query +SELECT id, array_min(array(array(a), array(b))), + array_min(array(named_struct('s', a, 'f', x), named_struct('s', b, 'f', y))) +FROM test_array_min_collation + +-- Lowercase ordering differs from binary ordering for the column values 'a' and 'B'. +query expect_fallback(Array extrema use binary string ordering) +SELECT id, array_min(array( + CAST(a AS STRING COLLATE UTF8_LCASE), + CAST(b AS STRING COLLATE UTF8_LCASE))) +FROM test_array_min_collation + +query expect_fallback(Array extrema use binary string ordering) +SELECT id, array_min(array( + array(CAST(a AS STRING COLLATE UTF8_LCASE)), + array(CAST(b AS STRING COLLATE UTF8_LCASE)))) +FROM test_array_min_collation + +-- A floating-point field does not make a collated struct eligible for native ordering. +query expect_fallback(Array extrema use binary string ordering) +SELECT id, array_min(array( + named_struct('s', CAST(a AS STRING COLLATE UTF8_LCASE), 'f', x), + named_struct('s', CAST(b AS STRING COLLATE UTF8_LCASE), 'f', y))) +FROM test_array_min_collation + +-- The collation check must recurse through both struct fields and nested array elements. +query expect_fallback(Array extrema use binary string ordering) +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))) +FROM test_array_min_collation diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_min_floating_point.sql b/spark/src/test/resources/sql-tests/expressions/array/array_min_floating_point.sql new file mode 100644 index 0000000000..0002b8f4b6 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_min_floating_point.sql @@ -0,0 +1,189 @@ +-- 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 retains the first equal extremum: zero signs compare equal, as do NaNs. +-- Require exact native results in both floating-point modes without the codegen dispatcher. +-- The dictionary matrix varies the writer setting, not a guarantee of dictionary pages. +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false +-- Config: spark.comet.expression.ArrayMin.allowIncompatible=false +-- ConfigMatrix: spark.comet.exec.strictFloatingPoint=false,true +-- ConfigMatrix: parquet.enable.dictionary=false,true + +statement +CREATE TABLE test_array_min_floating_point(id int, d array, f array) USING parquet + +statement +INSERT INTO test_array_min_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'), NULL), array(NULL, float('0.0'), float('-0.0'), NULL)), + (4, array(NULL, double('-0.0'), double('0.0'), NULL), array(NULL, float('-0.0'), float('0.0'), NULL)), + (5, array(double('-0.0')), array(float('-0.0'))), + (6, array(double('0.0')), array(float('0.0'))), + (7, array(), array()), + (8, array(NULL), array(NULL)), + (9, array(NULL, NULL), array(NULL, NULL)), + (10, NULL, NULL), + (11, array(double('-3.0'), double('1.0'), double('2.0')), array(float('-3.0'), float('1.0'), float('2.0'))), + (12, array(NULL, double('-2.0'), double('4.0'), NULL), array(NULL, float('-2.0'), float('4.0'), NULL)), + (13, array(double('NaN'), double('1.0'), double('2.0')), array(float('NaN'), float('1.0'), float('2.0'))), + (14, array(double('1.0'), double('NaN'), double('2.0')), array(float('1.0'), float('NaN'), float('2.0'))), + (15, array(double('1.0'), double('2.0'), double('NaN')), array(float('1.0'), float('2.0'), float('NaN'))), + (16, array(double('NaN'), double('NaN')), array(float('NaN'), float('NaN'))), + (17, array(NULL, double('NaN'), NULL), array(NULL, float('NaN'), NULL)), + (18, array(double('-Infinity'), double('Infinity'), double('NaN')), array(float('-Infinity'), float('Infinity'), float('NaN'))), + (19, array(double('NaN'), double('Infinity'), double('-Infinity')), array(float('NaN'), float('Infinity'), float('-Infinity'))), + (20, array(double('-Infinity'), double('1.0'), double('Infinity')), array(float('-Infinity'), float('1.0'), float('Infinity'))), + (21, array(double('Infinity'), double('1.0'), double('-Infinity')), array(float('Infinity'), float('1.0'), float('-Infinity'))), + (22, array(double('Infinity'), double('Infinity')), array(float('Infinity'), float('Infinity'))), + (23, array(double('-Infinity'), double('-Infinity')), array(float('-Infinity'), float('-Infinity'))) + +query +SELECT id, array_min(d), array_min(f) FROM test_array_min_floating_point + +-- The harness disables constant folding, so literal inputs exercise native evaluation too. +query +SELECT array_min(array(double('0.0'), double('-0.0'))), + array_min(array(double('-0.0'), double('0.0'))), + array_min(array(float('0.0'), float('-0.0'))), + array_min(array(float('-0.0'), float('0.0'))) + +query +SELECT array_min(CAST(NULL AS array)), + array_min(CAST(NULL AS array)), + array_min(CAST(array() AS array)), + array_min(CAST(array() AS array)), + array_min(array(CAST(NULL AS double), CAST(NULL AS double))), + array_min(array(CAST(NULL AS float), CAST(NULL AS float))) + +query +SELECT array_min(array(double('NaN'), double('Infinity'), double('-Infinity'))), + array_min(array(float('NaN'), float('Infinity'), float('-Infinity'))), + array_min(array(double('NaN'), double('NaN'))), + array_min(array(float('NaN'), float('NaN'))), + array_min(array(double('Infinity'), double('-Infinity'))), + array_min(array(float('Infinity'), float('-Infinity'))) + +-- Exercise both tie orders around 32-element boundaries and across longer arrays. +-- Build the input during INSERT so the tested projection has only column arguments. +statement +CREATE TABLE test_array_min_floating_long( + n int, negative_first boolean, d array, f array, + nullable_d array, nullable_f array) USING parquet + +statement +INSERT INTO test_array_min_floating_long +SELECT n, negative_first, + concat(array(d0), array_repeat(d1, n - 1)), + concat(array(f0), array_repeat(f1, n - 1)), + concat(array_repeat(CAST(NULL AS double), n - 2), array(d0, d1)), + concat(array_repeat(CAST(NULL AS float), n - 2), array(f0, f1)) +FROM VALUES (31), (32), (33), (65), (129) AS sizes(n) +CROSS JOIN VALUES + (false, double('0.0'), double('-0.0'), float('0.0'), float('-0.0')), + (true, double('-0.0'), double('0.0'), float('-0.0'), float('0.0')) +AS signs(negative_first, d0, d1, f0, f1) + +query +SELECT n, negative_first, array_min(d), array_min(f), + array_min(nullable_d), array_min(nullable_f) +FROM test_array_min_floating_long + +-- Outer null elements are skipped; nulls inside an array sort before non-null elements. +-- Equal nested zeros and NaNs must allow comparison to continue to later elements. +statement +CREATE TABLE test_array_min_floating_nested( + id int, d array>, f array>) USING parquet + +statement +INSERT INTO test_array_min_floating_nested VALUES + (1, array(array(double('0.0')), array(double('-0.0'))), + array(array(float('0.0')), array(float('-0.0')))), + (2, array(array(double('-0.0')), array(double('0.0'))), + array(array(float('-0.0')), array(float('0.0')))), + (3, array(array(NULL, double('0.0')), array(NULL, double('-0.0'))), + array(array(NULL, float('0.0')), array(NULL, float('-0.0')))), + (4, array(array(NULL, double('-0.0')), array(NULL, double('0.0'))), + array(array(NULL, float('-0.0')), array(NULL, float('0.0')))), + (5, array(array(NULL), array(double('-Infinity'))), + array(array(NULL), array(float('-Infinity')))), + (6, array(array(double('-Infinity')), array(NULL)), + array(array(float('-Infinity')), array(NULL))), + (7, array(array(double('1.0'), NULL), array(double('1.0'), double('2.0'))), + array(array(float('1.0'), NULL), array(float('1.0'), float('2.0')))), + (8, array(array(double('1.0')), array(double('1.0'), NULL)), + array(array(float('1.0')), array(float('1.0'), NULL))), + (9, array(array(), array(NULL)), array(array(), array(NULL))), + (10, array(NULL, array(double('-0.0')), array(double('0.0'))), + array(NULL, array(float('-0.0')), array(float('0.0')))), + (11, array(NULL, NULL), array(NULL, NULL)), + (12, array(), array()), + (13, NULL, NULL), + (14, array(array(double('NaN'), double('2.0')), array(double('NaN'), double('1.0'))), + array(array(float('NaN'), float('2.0')), array(float('NaN'), float('1.0')))), + (15, array(array(double('0.0'), double('1.0')), array(double('-0.0'), double('2.0'))), + array(array(float('0.0'), float('1.0')), array(float('-0.0'), float('2.0')))) + +query +SELECT id, array_min(d), array_min(f) FROM test_array_min_floating_nested + +-- A zero tie in the first struct field must not override the ordering of its payload. +-- When every field compares equal, preserve the first struct, including its zero sign. +statement +CREATE TABLE test_array_min_floating_struct( + id int, d array>, f array>) USING parquet + +statement +INSERT INTO test_array_min_floating_struct VALUES + (1, array(named_struct('v', double('0.0'), 'payload', 1), named_struct('v', double('-0.0'), 'payload', 2)), + array(named_struct('v', float('0.0'), 'payload', 1), named_struct('v', float('-0.0'), 'payload', 2))), + (2, array(named_struct('v', double('-0.0'), 'payload', 2), named_struct('v', double('0.0'), 'payload', 1)), + array(named_struct('v', float('-0.0'), 'payload', 2), named_struct('v', float('0.0'), 'payload', 1))), + (3, array(named_struct('v', double('0.0'), 'payload', 1), named_struct('v', double('-0.0'), 'payload', 1)), + array(named_struct('v', float('0.0'), 'payload', 1), named_struct('v', float('-0.0'), 'payload', 1))), + (4, array(named_struct('v', double('-0.0'), 'payload', 1), named_struct('v', double('0.0'), 'payload', 1)), + array(named_struct('v', float('-0.0'), 'payload', 1), named_struct('v', float('0.0'), 'payload', 1))), + (5, array(named_struct('v', NULL, 'payload', 2), named_struct('v', double('-Infinity'), 'payload', 1)), + array(named_struct('v', NULL, 'payload', 2), named_struct('v', float('-Infinity'), 'payload', 1))), + (6, array(named_struct('v', double('0.0'), 'payload', NULL), named_struct('v', double('-0.0'), 'payload', 1)), + array(named_struct('v', float('0.0'), 'payload', NULL), named_struct('v', float('-0.0'), 'payload', 1))), + (7, array(named_struct('v', double('NaN'), 'payload', 2), named_struct('v', double('NaN'), 'payload', 1)), + array(named_struct('v', float('NaN'), 'payload', 2), named_struct('v', float('NaN'), 'payload', 1))), + (8, array(NULL, named_struct('v', NULL, 'payload', NULL)), + array(NULL, named_struct('v', NULL, 'payload', NULL))), + (9, array(NULL, NULL), array(NULL, NULL)), + (10, array(), array()), + (11, NULL, NULL) + +query +SELECT id, array_min(d), array_min(f) FROM test_array_min_floating_struct + +-- Recurse through a list of structs, not only structs or lists of primitive elements. +query +SELECT id, array_min(array(array(d[0]), array(d[1]))), + array_min(array(array(f[0]), array(f[1]))) +FROM test_array_min_floating_struct WHERE id IN (1, 2, 3, 4) + +-- Recurse in the other direction too: an array-valued struct field tied on zero. +query +SELECT array_min(array( + named_struct('v', array(double('0.0')), 'payload', 1), + named_struct('v', array(double('-0.0')), 'payload', 2))), + array_min(array( + named_struct('v', array(float('0.0')), 'payload', 1), + named_struct('v', array(float('-0.0')), 'payload', 2))) diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index 05a6e8e650..4f4637fccd 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.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions._ @@ -560,6 +560,59 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp } } + 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)