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 a068e12a50..d650b49ed8 100644 --- a/docs/source/user-guide/latest/compatibility/floating-point.md +++ b/docs/source/user-guide/latest/compatibility/floating-point.md @@ -28,6 +28,13 @@ to Spark in some cases, especially when the data contains both positive and nega case that is not of concern for many users. If it is a concern, setting `spark.comet.exec.strictFloatingPoint=true` will make relevant operations fall back to Spark. +`array_min` and `array_max` use Spark-compatible native comparisons in both strict and non-strict +floating-point modes. Signed zeros compare equal, and all NaN representations compare equal and +greater than non-NaN values. The original first equal element is retained: for example, +`array_min(array(0.0D, -0.0D))` returns `0.0`, while reversing those elements returns `-0.0`. +The same ordering applies recursively to floating-point fields in arrays and structs. These +expressions do not require Spark's codegen dispatcher for floating-point compatibility. + ## Ordering: signed zero (`-0.0` vs `+0.0`) Spark's `ORDER BY`, `RANK`, `DENSE_RANK`, and window frame comparisons route through diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 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 b8c0394558..e0184ecc27 100644 --- a/native/spark-expr/Cargo.toml +++ b/native/spark-expr/Cargo.toml @@ -171,6 +171,10 @@ harness = false name = "arrays_overlap" harness = false +[[bench]] +name = "array_extrema" +harness = false + [[bench]] name = "checked_arithmetic" harness = false @@ -237,4 +241,4 @@ harness = false [[bench]] name = "abs" -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..b8316fe27c --- /dev/null +++ b/native/spark-expr/src/array_funcs/array_extrema.rs @@ -0,0 +1,425 @@ +// 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)?, + // The delegated return_type above rejects an outer FixedSizeList. Supporting it as + // a nested element in spark_comparator does not broaden the function's signature. + 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. +/// Checking only Arrow's winner cannot establish that its answer agrees with Spark: +/// for max([-NaN, 1.0]), Arrow's total ordering selects 1.0 while Spark selects NaN. +/// A corrective scan triggered only by a zero or NaN winner would miss that case. +fn float_extrema( + array: &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. + // Comparator construction happens once per child array/batch, not per comparison; + // the decoded array is retained by the comparator for the subsequent index scan. + 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 c68b998e61..4bc0616274 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}; @@ -285,6 +285,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)