From b72f556cd7916675375ad1bcf41ea7d243649341 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Thu, 20 Aug 2026 21:47:20 -0700 Subject: [PATCH 1/3] fix: honor strict floating-point mode for array extrema --- .../expression-audits/array_funcs.md | 20 +++---- .../latest/compatibility/floating-point.md | 8 +++ docs/source/user-guide/latest/expressions.md | 4 +- .../scala/org/apache/comet/GenerateDocs.scala | 5 +- .../scala/org/apache/comet/serde/arrays.scala | 28 +++++++++- .../array_extrema_strict_fp_fallback.sql | 50 +++++++++++++++++ .../array/array_extrema_strict_fp_opt_in.sql | 42 ++++++++++++++ .../sql-tests/expressions/array/array_max.sql | 12 ++++ .../expressions/array/array_max_strict_fp.sql | 55 +++++++++++++++++++ .../sql-tests/expressions/array/array_min.sql | 12 ++-- .../expressions/array/array_min_strict_fp.sql | 55 +++++++++++++++++++ 11 files changed, 269 insertions(+), 22 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_fallback.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_opt_in.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_max_strict_fp.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_min_strict_fp.sql diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index 31261b25427..4d0ce0504c1 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-20): identical to 3.5.8. +- Spark 3.5.8 (audited 2026-08-20): `ArrayMax` skips NULL elements and returns NULL for an empty or all-NULL array. `SQLOrderingUtil` treats NaN as greater than non-NaN values and signed zeros as equal. The first equal maximum is retained. +- Spark 4.0.1 (audited 2026-08-20): `NullIntolerant` becomes a `nullIntolerant` field. Extrema semantics are unchanged. +- Spark 4.1.1 (audited 2026-08-20): identical to 4.0.1. +- Current status: `CometArrayMax` delegates to DataFusion's `array_max`. The native path chooses `+0.0` for a signed-zero tie, unlike Spark when `-0.0` appears first. With `spark.comet.exec.strictFloatingPoint=true`, element types containing float/double are `Incompatible` and use Spark's codegen dispatcher unless native incompatibility is explicitly allowed. If the dispatcher is disabled, they fall back to Spark. Non-strict native behavior is unchanged. Native signed-zero parity remains tracked by [#5401](https://github.com/apache/datafusion-comet/issues/5401). ## 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-20): identical to 3.5.8. +- Spark 3.5.8 (audited 2026-08-20): mirrors `ArrayMax`, retaining the first equal minimum. The NULL, NaN, and signed-zero comparison rules are the same. +- Spark 4.0.1 (audited 2026-08-20): same trait refactor as `array_max`, with no change in extrema semantics. +- Spark 4.1.1 (audited 2026-08-20): identical to 4.0.1. +- Current status: `CometArrayMin` delegates to DataFusion's `array_min`. The native path chooses `-0.0` for a signed-zero tie, unlike Spark when `+0.0` appears first. Its strict-mode support level and codegen/fallback behavior match `array_max`. Non-strict native behavior is unchanged. Native signed-zero parity remains tracked by [#5401](https://github.com/apache/datafusion-comet/issues/5401). ## 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 ffab0550609..1774d8c58eb 100644 --- a/docs/source/user-guide/latest/compatibility/floating-point.md +++ b/docs/source/user-guide/latest/compatibility/floating-point.md @@ -27,3 +27,11 @@ So Comet adds additional normalization expression of NaN and zero for comparison to Spark in some cases, especially when the data contains both positive and negative zero. This is likely an edge case that is not of concern for many users. If it is a concern, setting `spark.comet.exec.strictFloatingPoint=true` will make relevant operations fall back to Spark. + +For `array_min` and `array_max`, Spark retains the first element when signed zeros compare equal. +The native implementation instead orders `-0.0` before `+0.0`, so a tied minimum can return `-0.0` +where Spark returns `+0.0`, and a tied maximum can return `+0.0` where Spark returns `-0.0`. +With `spark.comet.exec.strictFloatingPoint=true`, these expressions use Spark's codegen dispatcher +inside Comet, or fall back to Spark if that dispatcher is disabled. The native path remains available +when the expression's `allowIncompatible` setting is explicitly enabled. Non-strict behavior is +unchanged. Native signed-zero parity is tracked by [#5401](https://github.com/apache/datafusion-comet/issues/5401). diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 192eae807eb..aaf202e76ce 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` | ✅ | Hybrid | Signed-zero ties may differ natively; strict floating-point mode uses Spark's codegen dispatcher ([details](compatibility/floating-point.md)) | +| `array_min` | ✅ | Hybrid | Signed-zero ties may differ natively; strict floating-point mode uses Spark's codegen dispatcher ([details](compatibility/floating-point.md)) | | `array_position` | ✅ | Native | Binary/struct/map/null elements fall back | | `array_prepend` | ✅ | — | | | `array_remove` | ✅ | Native | | diff --git a/spark/src/main/scala/org/apache/comet/GenerateDocs.scala b/spark/src/main/scala/org/apache/comet/GenerateDocs.scala index e5d0c942154..e16243ee30c 100644 --- a/spark/src/main/scala/org/apache/comet/GenerateDocs.scala +++ b/spark/src/main/scala/org/apache/comet/GenerateDocs.scala @@ -53,7 +53,7 @@ object GenerateDocs { * @param unsupportedReasons * cases that Comet's native implementation does not handle * @param nativeOptIn - * whether the serde implements `NativeOptInAvailable`, meaning the expression runs a + * whether the serde implements `NativeOptInAvailable`, meaning incompatible cases run a * Spark-compatible path by default and the user can opt into a native path * @param nativeOptInConfigKey * the config key the user sets to opt into the native path @@ -398,7 +398,8 @@ object GenerateDocs { } if (n.incompatibleReasons.nonEmpty) { val header = if (n.nativeOptIn) { - s"\nBy default, `$name` is evaluated in the JVM using Spark's own code-generated" + + s"\nFor the incompatible cases listed below, `$name` is evaluated by default" + + " in the JVM using Spark's own code-generated" + " implementation (run inside the Comet pipeline), which matches Spark exactly." + s" Set `${n.nativeOptInConfigKey}=true` to opt into Comet's native implementation" + " instead, which has the following differences from Spark:\n\n" diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala b/spark/src/main/scala/org/apache/comet/serde/arrays.scala index 748b1cee231..3b91b64284a 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 } } -object CometArrayMax extends CometExpressionSerde[ArrayMax] { +private object ArrayExtremaSupport { + val incompatReason: String = + s"With `${CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key}=true`, floating-point array " + + "extrema are incompatible because native comparisons order `-0.0` before `+0.0`, " + + "while Spark preserves the first equal element " + + "([#5401](https://github.com/apache/datafusion-comet/issues/5401))." + + def getSupportLevel(elementType: DataType): SupportLevel = + SupportLevel + .strictFloatingPointReason(elementType, "Finding floating-point array extrema") + .map(_ => Incompatible(Some(incompatReason))) + .getOrElse(Compatible()) +} + +object CometArrayMax extends CometExpressionSerde[ArrayMax] with CodegenDispatchFallback { + 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], @@ -228,7 +247,12 @@ object CometArrayMax extends CometExpressionSerde[ArrayMax] { } } -object CometArrayMin extends CometExpressionSerde[ArrayMin] { +object CometArrayMin extends CometExpressionSerde[ArrayMin] with CodegenDispatchFallback { + override def getIncompatibleReasons(): Seq[String] = Seq(ArrayExtremaSupport.incompatReason) + + override def getSupportLevel(expr: ArrayMin): SupportLevel = + ArrayExtremaSupport.getSupportLevel(expr.dataType) + override def convert( expr: ArrayMin, inputs: Seq[Attribute], diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_fallback.sql b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_fallback.sql new file mode 100644 index 00000000000..56f40fc6b7d --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_fallback.sql @@ -0,0 +1,50 @@ +-- 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. + +-- Disabling the codegen dispatcher must not send incompatible floating-point extrema +-- back to the native UDF. Non-floating extrema still have a native path. +-- https://github.com/apache/datafusion-comet/issues/5401 + +-- Config: spark.comet.exec.strictFloatingPoint=true +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false +-- Config: spark.comet.expression.ArrayMin.allowIncompatible=false +-- Config: spark.comet.expression.ArrayMax.allowIncompatible=false +-- ConfigMatrix: parquet.enable.dictionary=false,true + +statement +CREATE TABLE test_array_extrema_strict_fallback(id int, d array, f array, a int, b int) USING parquet + +statement +INSERT INTO test_array_extrema_strict_fallback VALUES + (1, array(double('0.0'), double('-0.0')), array(float('0.0'), float('-0.0')), 1, -2), + (2, array(double('-0.0'), double('0.0')), array(float('-0.0'), float('0.0')), -3, 4), + (3, array(), array(), NULL, 5), + (4, NULL, NULL, NULL, NULL) + +query expect_fallback(spark.comet.exec.strictFloatingPoint=true) +SELECT id, array_min(d), array_min(f) FROM test_array_extrema_strict_fallback + +query expect_fallback(spark.comet.exec.strictFloatingPoint=true) +SELECT id, array_max(d), array_max(f) FROM test_array_extrema_strict_fallback + +query expect_fallback(spark.comet.exec.strictFloatingPoint=true) +SELECT id, array_min(array(array(d[0]), array(d[1]))), + array_max(array(array(f[0]), array(f[1]))) +FROM test_array_extrema_strict_fallback WHERE id IN (1, 2) + +query +SELECT id, array_min(array(a, b)), array_max(array(a, b)) FROM test_array_extrema_strict_fallback diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_opt_in.sql b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_opt_in.sql new file mode 100644 index 00000000000..4617ff2a353 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_opt_in.sql @@ -0,0 +1,42 @@ +-- 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. + +-- Explicitly allowing incompatible extrema must retain native execution in strict mode, +-- even with the codegen dispatcher disabled. These arrays have no signed-zero ties, whose +-- native parity remains tracked by https://github.com/apache/datafusion-comet/issues/5401. + +-- Config: spark.comet.exec.strictFloatingPoint=true +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false +-- Config: spark.comet.expression.ArrayMin.allowIncompatible=true +-- Config: spark.comet.expression.ArrayMax.allowIncompatible=true +-- ConfigMatrix: parquet.enable.dictionary=false,true + +statement +CREATE TABLE test_array_extrema_strict_opt_in(id int, d array, f array) USING parquet + +statement +INSERT INTO test_array_extrema_strict_opt_in VALUES + (1, array(double('-3.0'), double('1.0'), double('2.0')), array(float('-3.0'), float('1.0'), float('2.0'))), + (2, array(double('0.0')), array(float('0.0'))), + (3, array(double('-0.0')), array(float('-0.0'))), + (4, array(NULL, double('-2.0'), double('4.0')), array(NULL, float('-2.0'), float('4.0'))), + (5, array(), array()), + (6, NULL, NULL) + +query +SELECT id, array_min(d), array_max(d), array_min(f), array_max(f) +FROM test_array_extrema_strict_opt_in 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 e5f9db3e8a4..a84e41314f3 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 @@ -68,3 +68,15 @@ INSERT INTO test_array_max_float VALUES query SELECT array_max(arr) FROM test_array_max_float + +-- Spark preserves the first equal zero (-0.0 here); the native maximum returns +0.0. +-- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. +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 ignore(https://github.com/apache/datafusion-comet/issues/5401) +SELECT array_max(d), array_max(f) FROM test_array_max_negzero diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_max_strict_fp.sql b/spark/src/test/resources/sql-tests/expressions/array/array_max_strict_fp.sql new file mode 100644 index 00000000000..281894dba2e --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_max_strict_fp.sql @@ -0,0 +1,55 @@ +-- 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. + +-- Strict mode uses Spark's codegen for signed-zero ties. Spark retains the first equal +-- extremum, so both input orders and floating-point widths must be checked without a tolerance. +-- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. + +-- Config: spark.comet.exec.strictFloatingPoint=true +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true +-- Config: spark.comet.expression.ArrayMax.allowIncompatible=false +-- ConfigMatrix: parquet.enable.dictionary=false,true + +statement +CREATE TABLE test_array_max_strict_fp(id int, d array, f array) USING parquet + +statement +INSERT INTO test_array_max_strict_fp VALUES + (1, array(double('0.0'), double('-0.0')), array(float('0.0'), float('-0.0'))), + (2, array(double('-0.0'), double('0.0')), array(float('-0.0'), float('0.0'))), + (3, array(NULL, double('0.0'), double('-0.0')), array(NULL, float('0.0'), float('-0.0'))), + (4, array(NULL, double('-0.0'), double('0.0')), array(NULL, float('-0.0'), float('0.0'))), + (5, array(double('-0.0')), array(float('-0.0'))), + (6, array(), array()), + (7, array(NULL), array(NULL)), + (8, NULL, NULL) + +query +SELECT id, array_max(d), array_max(f) FROM test_array_max_strict_fp + +-- The result type contains a float/double inside an array, so the strict-mode guard is recursive. +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_strict_fp WHERE id IN (1, 2) + +-- The SQL harness disables constant folding, so literal inputs exercise the dispatcher 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'))) 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 f3efb870aba..fe825675f42 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 @@ -49,8 +49,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 +-- Spark preserves the first equal zero (+0.0 here); the native minimum returns -0.0. +-- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. statement CREATE TABLE test_array_min_dbl_negzero(arr array) USING parquet @@ -58,7 +58,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 ignore(https://github.com/apache/datafusion-comet/issues/5401) SELECT array_min(arr) FROM test_array_min_dbl_negzero -- ===== FLOAT arrays with NaN/Infinity/-0.0 ===== @@ -79,8 +79,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 +-- Spark preserves the first equal zero (+0.0 here); the native minimum returns -0.0. +-- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. statement CREATE TABLE test_array_min_flt_negzero(arr array) USING parquet @@ -88,5 +88,5 @@ statement INSERT INTO test_array_min_flt_negzero VALUES (array(CAST(0.0 AS FLOAT), float('-0.0'))) -query ignore(array_min signed-zero: Spark +0.0, Comet -0.0) +query ignore(https://github.com/apache/datafusion-comet/issues/5401) SELECT array_min(arr) FROM test_array_min_flt_negzero diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_min_strict_fp.sql b/spark/src/test/resources/sql-tests/expressions/array/array_min_strict_fp.sql new file mode 100644 index 00000000000..f0106b0e3d0 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_min_strict_fp.sql @@ -0,0 +1,55 @@ +-- 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. + +-- Strict mode uses Spark's codegen for signed-zero ties. Spark retains the first equal +-- extremum, so both input orders and floating-point widths must be checked without a tolerance. +-- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. + +-- Config: spark.comet.exec.strictFloatingPoint=true +-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true +-- Config: spark.comet.expression.ArrayMin.allowIncompatible=false +-- ConfigMatrix: parquet.enable.dictionary=false,true + +statement +CREATE TABLE test_array_min_strict_fp(id int, d array, f array) USING parquet + +statement +INSERT INTO test_array_min_strict_fp VALUES + (1, array(double('0.0'), double('-0.0')), array(float('0.0'), float('-0.0'))), + (2, array(double('-0.0'), double('0.0')), array(float('-0.0'), float('0.0'))), + (3, array(NULL, double('0.0'), double('-0.0')), array(NULL, float('0.0'), float('-0.0'))), + (4, array(NULL, double('-0.0'), double('0.0')), array(NULL, float('-0.0'), float('0.0'))), + (5, array(double('-0.0')), array(float('-0.0'))), + (6, array(), array()), + (7, array(NULL), array(NULL)), + (8, NULL, NULL) + +query +SELECT id, array_min(d), array_min(f) FROM test_array_min_strict_fp + +-- The result type contains a float/double inside an array, so the strict-mode guard is recursive. +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_strict_fp WHERE id IN (1, 2) + +-- The SQL harness disables constant folding, so literal inputs exercise the dispatcher 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'))) From 79040a1bff89cd75cd8eea25d89bc307d5864ce3 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Sat, 22 Aug 2026 11:47:04 -0700 Subject: [PATCH 2/3] fix: match Spark ordering in native array extrema --- .../expression-audits/array_funcs.md | 20 +- .../latest/compatibility/floating-point.md | 13 +- docs/source/user-guide/latest/expressions.md | 4 +- native/spark-expr/Cargo.toml | 6 +- native/spark-expr/benches/array_extrema.rs | 187 +++ .../src/array_funcs/array_extrema.rs | 349 ++++++ .../src/array_funcs/array_extrema/tests.rs | 1042 +++++++++++++++++ native/spark-expr/src/array_funcs/mod.rs | 2 + native/spark-expr/src/comet_scalar_funcs.rs | 8 +- .../scala/org/apache/comet/GenerateDocs.scala | 5 +- .../scala/org/apache/comet/serde/arrays.scala | 26 +- .../array_extrema_strict_fp_fallback.sql | 50 - .../array/array_extrema_strict_fp_opt_in.sql | 42 - .../sql-tests/expressions/array/array_max.sql | 40 +- .../expressions/array/array_max_collation.sql | 72 ++ .../array/array_max_floating_point.sql | 189 +++ .../expressions/array/array_max_strict_fp.sql | 55 - .../sql-tests/expressions/array/array_min.sql | 46 +- .../expressions/array/array_min_collation.sql | 72 ++ .../array/array_min_floating_point.sql | 189 +++ .../expressions/array/array_min_strict_fp.sql | 55 - .../comet/CometArrayExpressionSuite.scala | 55 +- 22 files changed, 2274 insertions(+), 253 deletions(-) create mode 100644 native/spark-expr/benches/array_extrema.rs create mode 100644 native/spark-expr/src/array_funcs/array_extrema.rs create mode 100644 native/spark-expr/src/array_funcs/array_extrema/tests.rs delete mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_fallback.sql delete mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_opt_in.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_max_collation.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_max_floating_point.sql delete mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_max_strict_fp.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_min_collation.sql create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_min_floating_point.sql delete mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_min_strict_fp.sql diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index 4d0ce0504c1..08f5c0d1acb 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-08-20): identical to 3.5.8. -- Spark 3.5.8 (audited 2026-08-20): `ArrayMax` skips NULL elements and returns NULL for an empty or all-NULL array. `SQLOrderingUtil` treats NaN as greater than non-NaN values and signed zeros as equal. The first equal maximum is retained. -- Spark 4.0.1 (audited 2026-08-20): `NullIntolerant` becomes a `nullIntolerant` field. Extrema semantics are unchanged. -- Spark 4.1.1 (audited 2026-08-20): identical to 4.0.1. -- Current status: `CometArrayMax` delegates to DataFusion's `array_max`. The native path chooses `+0.0` for a signed-zero tie, unlike Spark when `-0.0` appears first. With `spark.comet.exec.strictFloatingPoint=true`, element types containing float/double are `Incompatible` and use Spark's codegen dispatcher unless native incompatibility is explicitly allowed. If the dispatcher is disabled, they fall back to Spark. Non-strict native behavior is unchanged. Native signed-zero parity remains tracked by [#5401](https://github.com/apache/datafusion-comet/issues/5401). +- 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-08-20): identical to 3.5.8. -- Spark 3.5.8 (audited 2026-08-20): mirrors `ArrayMax`, retaining the first equal minimum. The NULL, NaN, and signed-zero comparison rules are the same. -- Spark 4.0.1 (audited 2026-08-20): same trait refactor as `array_max`, with no change in extrema semantics. -- Spark 4.1.1 (audited 2026-08-20): identical to 4.0.1. -- Current status: `CometArrayMin` delegates to DataFusion's `array_min`. The native path chooses `-0.0` for a signed-zero tie, unlike Spark when `+0.0` appears first. Its strict-mode support level and codegen/fallback behavior match `array_max`. Non-strict native behavior is unchanged. Native signed-zero parity remains tracked by [#5401](https://github.com/apache/datafusion-comet/issues/5401). +- 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 1774d8c58eb..edee0a487d3 100644 --- a/docs/source/user-guide/latest/compatibility/floating-point.md +++ b/docs/source/user-guide/latest/compatibility/floating-point.md @@ -28,10 +28,9 @@ 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. -For `array_min` and `array_max`, Spark retains the first element when signed zeros compare equal. -The native implementation instead orders `-0.0` before `+0.0`, so a tied minimum can return `-0.0` -where Spark returns `+0.0`, and a tied maximum can return `+0.0` where Spark returns `-0.0`. -With `spark.comet.exec.strictFloatingPoint=true`, these expressions use Spark's codegen dispatcher -inside Comet, or fall back to Spark if that dispatcher is disabled. The native path remains available -when the expression's `allowIncompatible` setting is explicitly enabled. Non-strict behavior is -unchanged. Native signed-zero parity is tracked by [#5401](https://github.com/apache/datafusion-comet/issues/5401). +`array_min` and `array_max` use Spark-compatible native comparisons in both strict and non-strict +floating-point modes. Signed zeros compare equal, and all NaN representations compare equal and +greater than non-NaN values. The original first equal element is retained: for example, +`array_min(array(0.0D, -0.0D))` returns `0.0`, while reversing those elements returns `-0.0`. +The same ordering applies recursively to floating-point fields in arrays and structs. These +expressions do not require Spark's codegen dispatcher for floating-point compatibility. diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index aaf202e76ce..c71eb96e8b1 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` | ✅ | Hybrid | Signed-zero ties may differ natively; strict floating-point mode uses Spark's codegen dispatcher ([details](compatibility/floating-point.md)) | -| `array_min` | ✅ | Hybrid | Signed-zero ties may differ natively; strict floating-point mode uses Spark's codegen dispatcher ([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 6faa9fec4ec..beee7bde19a 100644 --- a/native/spark-expr/Cargo.toml +++ b/native/spark-expr/Cargo.toml @@ -168,6 +168,10 @@ harness = false name = "arrays_overlap" harness = false +[[bench]] +name = "array_extrema" +harness = false + [[bench]] name = "checked_arithmetic" harness = false @@ -222,4 +226,4 @@ harness = false [[bench]] name = "cast_int_to_decimal" -harness = false \ No newline at end of file +harness = false diff --git a/native/spark-expr/benches/array_extrema.rs b/native/spark-expr/benches/array_extrema.rs new file mode 100644 index 00000000000..0870b077139 --- /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 00000000000..d9858ac165e --- /dev/null +++ b/native/spark-expr/src/array_funcs/array_extrema.rs @@ -0,0 +1,349 @@ +// 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_comparator, new_empty_array, Array, ArrayRef, AsArray, DynComparator, GenericListArray, + GenericListViewArray, OffsetSizeTrait, PrimitiveArray, PrimitiveBuilder, UInt64Array, +}; +use arrow::buffer::NullBuffer; +use arrow::compute::{cast, take, SortOptions}; +use arrow::datatypes::{ArrowPrimitiveType, DataType, Float32Type, Float64Type}; +use datafusion::common::{exec_err, Result, ScalarValue}; +use datafusion::functions_nested::min_max::{array_max_udf, array_min_udf}; +use datafusion::logical_expr::{ + ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, +}; +use num::Float; + +/// Spark's array_min/array_max retain the first non-null value on an ordering tie. +/// In particular, signed zeros compare equal and all NaNs compare equal and greater +/// than non-NaNs. Nested arrays and structs use the same ordering, with nulls first. +#[derive(Debug, Hash, Eq, PartialEq)] +pub struct SparkArrayExtrema { + is_min: bool, + datafusion_udf: Arc, +} + +impl SparkArrayExtrema { + pub fn new(is_min: bool) -> Self { + Self { + is_min, + // Capture the original implementation, not a registry lookup: these UDFs + // replace the DataFusion names in Comet's function registry. + datafusion_udf: if is_min { + array_min_udf() + } else { + array_max_udf() + }, + } + } +} + +impl ScalarUDFImpl for SparkArrayExtrema { + fn name(&self) -> &str { + if self.is_min { + "array_min" + } else { + "array_max" + } + } + + fn signature(&self) -> &Signature { + self.datafusion_udf.signature() + } + + fn return_type(&self, arg_types: &[DataType]) -> Result { + self.datafusion_udf.return_type(arg_types) + } + + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { + let [input] = args.args.as_slice() else { + return exec_err!("{} takes exactly one argument", self.name()); + }; + let element_type = self.return_type(&[input.data_type()])?; + + // DataFusion's non-primitive path reconstructs an array from scalars, which + // cannot infer a type from an empty iterator. Keep the declared element type. + if matches!(input, ColumnarValue::Array(array) if array.is_empty()) { + return Ok(ColumnarValue::Array(new_empty_array(&element_type))); + } + if !needs_spark_ordering(&element_type) { + return self.datafusion_udf.invoke_with_args(args); + } + + let is_scalar = matches!(input, ColumnarValue::Scalar(_)); + let array = match input { + ColumnarValue::Array(array) => Arc::clone(array), + ColumnarValue::Scalar(value) => value.to_array()?, + }; + let result = match array.data_type() { + DataType::List(_) => array_extrema(array.as_list::(), self.is_min)?, + DataType::LargeList(_) => array_extrema(array.as_list::(), self.is_min)?, + other => return exec_err!("{} does not support type {other}", self.name()), + }; + + if is_scalar { + Ok(ColumnarValue::Scalar(ScalarValue::try_from_array( + &result, 0, + )?)) + } else { + Ok(ColumnarValue::Array(result)) + } + } +} + +fn needs_spark_ordering(data_type: &DataType) -> bool { + matches!( + data_type, + DataType::Float32 + | DataType::Float64 + | DataType::List(_) + | DataType::LargeList(_) + | DataType::FixedSizeList(_, _) + | DataType::ListView(_) + | DataType::LargeListView(_) + | DataType::Struct(_) + | DataType::Dictionary(_, _) + ) +} + +fn array_extrema( + array: &GenericListArray, + is_min: bool, +) -> Result { + match array.value_type() { + DataType::Float32 => Ok(Arc::new(float_extrema::(array, is_min))), + DataType::Float64 => Ok(Arc::new(float_extrema::(array, is_min))), + _ => nested_extrema(array, is_min), + } +} + +/// Scan the flat value buffer for every list length. Arrow's float min/max kernels +/// use a different ordering, so long lists must not switch to those kernels. +fn float_extrema( + array: &GenericListArray, + is_min: bool, +) -> PrimitiveArray +where + T::Native: Float, +{ + let values = array.values().as_primitive::(); + let buffer = values.values(); + let nulls = values.nulls(); + let mut result = PrimitiveBuilder::::with_capacity(array.len()); + for (row, offsets) in array.offsets().windows(2).enumerate() { + let mut best: Option = None; + if array.is_valid(row) { + let start = offsets[0].as_usize(); + let end = offsets[1].as_usize(); + for (index, &candidate) in buffer[start..end].iter().enumerate() { + if nulls.is_some_and(|nulls| nulls.is_null(start + index)) { + continue; + } + let replace = match best { + None => true, + Some(current) if is_min => { + candidate < current || (!candidate.is_nan() && current.is_nan()) + } + Some(current) => { + candidate > current || (candidate.is_nan() && !current.is_nan()) + } + }; + if replace { + // Copy the winning value, never normalize its zero sign or NaN bits. + best = Some(candidate); + } + } + } + result.append_option(best); + } + result.finish() +} + +fn nested_extrema( + array: &GenericListArray, + is_min: bool, +) -> Result { + let values = array.values(); + let compare = spark_comparator(values)?; + // Dictionary keys can refer to null values even when the keys themselves are valid. + let nulls = values.logical_nulls(); + let ordering = if is_min { + Ordering::Less + } else { + Ordering::Greater + }; + let mut indices = Vec::with_capacity(array.len()); + for (row, offsets) in array.offsets().windows(2).enumerate() { + let mut best = None; + if array.is_valid(row) { + for candidate in offsets[0].as_usize()..offsets[1].as_usize() { + if nulls.as_ref().is_some_and(|nulls| nulls.is_null(candidate)) { + continue; + } + if best.is_none_or(|current| compare(candidate, current) == ordering) { + best = Some(candidate); + } + } + } + indices.push(best.map(|index| index as u64)); + } + // Take from the original values, not comparator-normalized or reconstructed values. + // This preserves nested fields, dictionary types, signed zeros, and NaN payloads. + Ok(take(values.as_ref(), &UInt64Array::from(indices), None)?) +} + +/// Build one comparator per child array, not per row. This is local to extrema: +/// DataFusion's ScalarValue nested comparisons put inner nulls last, unlike Spark. +fn spark_comparator(array: &ArrayRef) -> Result { + match array.data_type() { + DataType::Float32 => Ok(float_comparator::(array)), + DataType::Float64 => Ok(float_comparator::(array)), + DataType::List(_) => list_comparator(array.as_list::()), + DataType::LargeList(_) => list_comparator(array.as_list::()), + DataType::ListView(_) => list_view_comparator(array.as_list_view::()), + DataType::LargeListView(_) => list_view_comparator(array.as_list_view::()), + DataType::FixedSizeList(_, _) => { + let array = array.as_fixed_size_list(); + let compare = spark_comparator(array.values())?; + let size = array.value_length() as usize; + Ok(nulls_first(array.logical_nulls(), move |left, right| { + compare_ranges(left * size, size, right * size, size, &compare) + })) + } + DataType::Struct(_) => { + let array = array.as_struct(); + let fields = array + .columns() + .iter() + .map(spark_comparator) + .collect::>>()?; + Ok(nulls_first(array.logical_nulls(), move |left, right| { + fields + .iter() + .map(|compare| compare(left, right)) + .find(|&ordering| ordering != Ordering::Equal) + .unwrap_or(Ordering::Equal) + })) + } + DataType::Dictionary(_, value_type) => { + // Decode only for comparisons. Recursing after decoding also covers + // dictionaries whose values are nested arrays or structs with floats. + spark_comparator(&cast(array.as_ref(), value_type)?) + } + _ => Ok(make_comparator( + array.as_ref(), + array.as_ref(), + SortOptions { + descending: false, + nulls_first: true, + }, + )?), + } +} + +fn float_comparator(array: &ArrayRef) -> DynComparator +where + T::Native: Float, +{ + let values = array.as_primitive::().values().clone(); + nulls_first(array.logical_nulls(), move |left, right| { + let left = values[left]; + let right = values[right]; + if left == right || (left.is_nan() && right.is_nan()) { + Ordering::Equal + } else if left > right || left.is_nan() { + Ordering::Greater + } else { + Ordering::Less + } + }) +} + +fn list_comparator(array: &GenericListArray) -> Result { + let compare = spark_comparator(array.values())?; + let offsets = array.offsets().clone(); + Ok(nulls_first(array.logical_nulls(), move |left, right| { + let left_start = offsets[left].as_usize(); + let right_start = offsets[right].as_usize(); + compare_ranges( + left_start, + offsets[left + 1].as_usize() - left_start, + right_start, + offsets[right + 1].as_usize() - right_start, + &compare, + ) + })) +} + +fn list_view_comparator( + array: &GenericListViewArray, +) -> Result { + let compare = spark_comparator(array.values())?; + let offsets = array.offsets().clone(); + let sizes = array.sizes().clone(); + Ok(nulls_first(array.logical_nulls(), move |left, right| { + compare_ranges( + offsets[left].as_usize(), + sizes[left].as_usize(), + offsets[right].as_usize(), + sizes[right].as_usize(), + &compare, + ) + })) +} + +fn compare_ranges( + left_start: usize, + left_len: usize, + right_start: usize, + right_len: usize, + compare: &DynComparator, +) -> Ordering { + for offset in 0..left_len.min(right_len) { + let ordering = compare(left_start + offset, right_start + offset); + if ordering != Ordering::Equal { + return ordering; + } + } + left_len.cmp(&right_len) +} + +fn nulls_first( + nulls: Option, + compare: impl Fn(usize, usize) -> Ordering + Send + Sync + 'static, +) -> DynComparator { + match nulls { + None => Box::new(compare), + Some(nulls) => { + Box::new( + move |left, right| match (nulls.is_null(left), nulls.is_null(right)) { + (true, true) => Ordering::Equal, + (true, false) => Ordering::Less, + (false, true) => Ordering::Greater, + (false, false) => compare(left, right), + }, + ) + } + } +} + +#[cfg(test)] +mod tests; diff --git a/native/spark-expr/src/array_funcs/array_extrema/tests.rs b/native/spark-expr/src/array_funcs/array_extrema/tests.rs new file mode 100644 index 00000000000..9521d317c04 --- /dev/null +++ b/native/spark-expr/src/array_funcs/array_extrema/tests.rs @@ -0,0 +1,1042 @@ +// 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)), + } + } + } +} + +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 0c2c68dc6d9..25b1f5fcb67 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 dcb6b1906ce..90fdb26ad48 100644 --- a/native/spark-expr/src/comet_scalar_funcs.rs +++ b/native/spark-expr/src/comet_scalar_funcs.rs @@ -27,9 +27,9 @@ use crate::{ spark_ceil, spark_day_name, spark_decimal_div, spark_decimal_integral_div, spark_floor, spark_isnan, spark_lpad, spark_make_decimal, spark_month_name, spark_read_side_padding, spark_round, spark_rpad, spark_to_time, spark_unhex, spark_unscaled_value, EvalMode, - SparkArrayPositionFunc, SparkArraySlice, SparkArraysOverlap, SparkContains, SparkDateDiff, - SparkDateFromUnixDate, SparkDateTrunc, SparkFlatten, SparkMakeDate, SparkMakeInterval, - SparkMakeTime, SparkNextDay, SparkSecondsToTimestamp, SparkSizeFunc, + SparkArrayExtrema, SparkArrayPositionFunc, SparkArraySlice, SparkArraysOverlap, SparkContains, + SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc, SparkFlatten, SparkMakeDate, + SparkMakeInterval, SparkMakeTime, SparkNextDay, SparkSecondsToTimestamp, SparkSizeFunc, }; use arrow::datatypes::DataType; use datafusion::common::{DataFusionError, Result as DataFusionResult}; @@ -281,6 +281,8 @@ pub fn create_comet_physical_fun_with_eval_mode( fn all_scalar_functions() -> Vec> { vec![ + Arc::new(ScalarUDF::new_from_impl(SparkArrayExtrema::new(true))), + Arc::new(ScalarUDF::new_from_impl(SparkArrayExtrema::new(false))), Arc::new(ScalarUDF::new_from_impl(SparkArrayPositionFunc::default())), Arc::new(ScalarUDF::new_from_impl(SparkArraySlice::default())), Arc::new(ScalarUDF::new_from_impl(SparkArraysOverlap::default())), diff --git a/spark/src/main/scala/org/apache/comet/GenerateDocs.scala b/spark/src/main/scala/org/apache/comet/GenerateDocs.scala index e16243ee30c..e5d0c942154 100644 --- a/spark/src/main/scala/org/apache/comet/GenerateDocs.scala +++ b/spark/src/main/scala/org/apache/comet/GenerateDocs.scala @@ -53,7 +53,7 @@ object GenerateDocs { * @param unsupportedReasons * cases that Comet's native implementation does not handle * @param nativeOptIn - * whether the serde implements `NativeOptInAvailable`, meaning incompatible cases run a + * whether the serde implements `NativeOptInAvailable`, meaning the expression runs a * Spark-compatible path by default and the user can opt into a native path * @param nativeOptInConfigKey * the config key the user sets to opt into the native path @@ -398,8 +398,7 @@ object GenerateDocs { } if (n.incompatibleReasons.nonEmpty) { val header = if (n.nativeOptIn) { - s"\nFor the incompatible cases listed below, `$name` is evaluated by default" + - " in the JVM using Spark's own code-generated" + + s"\nBy default, `$name` is evaluated in the JVM using Spark's own code-generated" + " implementation (run inside the Comet pipeline), which matches Spark exactly." + s" Set `${n.nativeOptInConfigKey}=true` to opt into Comet's native implementation" + " instead, which has the following differences from Spark:\n\n" 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 3b91b64284a..cd9414a32b1 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -215,21 +215,21 @@ object CometArrayIntersect } } -private object ArrayExtremaSupport { +private object ArrayExtremaSupport extends CometTypeShim { val incompatReason: String = - s"With `${CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key}=true`, floating-point array " + - "extrema are incompatible because native comparisons order `-0.0` before `+0.0`, " + - "while Spark preserves the first equal element " + - "([#5401](https://github.com/apache/datafusion-comet/issues/5401))." - - def getSupportLevel(elementType: DataType): SupportLevel = - SupportLevel - .strictFloatingPointReason(elementType, "Finding floating-point array extrema") - .map(_ => Incompatible(Some(incompatReason))) - .getOrElse(Compatible()) + "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] with CodegenDispatchFallback { +object CometArrayMax extends CometExpressionSerde[ArrayMax] { override def getIncompatibleReasons(): Seq[String] = Seq(ArrayExtremaSupport.incompatReason) override def getSupportLevel(expr: ArrayMax): SupportLevel = @@ -247,7 +247,7 @@ object CometArrayMax extends CometExpressionSerde[ArrayMax] with CodegenDispatch } } -object CometArrayMin extends CometExpressionSerde[ArrayMin] with CodegenDispatchFallback { +object CometArrayMin extends CometExpressionSerde[ArrayMin] { override def getIncompatibleReasons(): Seq[String] = Seq(ArrayExtremaSupport.incompatReason) override def getSupportLevel(expr: ArrayMin): SupportLevel = diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_fallback.sql b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_fallback.sql deleted file mode 100644 index 56f40fc6b7d..00000000000 --- a/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_fallback.sql +++ /dev/null @@ -1,50 +0,0 @@ --- 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. - --- Disabling the codegen dispatcher must not send incompatible floating-point extrema --- back to the native UDF. Non-floating extrema still have a native path. --- https://github.com/apache/datafusion-comet/issues/5401 - --- Config: spark.comet.exec.strictFloatingPoint=true --- Config: spark.comet.exec.scalaUDF.codegen.enabled=false --- Config: spark.comet.expression.ArrayMin.allowIncompatible=false --- Config: spark.comet.expression.ArrayMax.allowIncompatible=false --- ConfigMatrix: parquet.enable.dictionary=false,true - -statement -CREATE TABLE test_array_extrema_strict_fallback(id int, d array, f array, a int, b int) USING parquet - -statement -INSERT INTO test_array_extrema_strict_fallback VALUES - (1, array(double('0.0'), double('-0.0')), array(float('0.0'), float('-0.0')), 1, -2), - (2, array(double('-0.0'), double('0.0')), array(float('-0.0'), float('0.0')), -3, 4), - (3, array(), array(), NULL, 5), - (4, NULL, NULL, NULL, NULL) - -query expect_fallback(spark.comet.exec.strictFloatingPoint=true) -SELECT id, array_min(d), array_min(f) FROM test_array_extrema_strict_fallback - -query expect_fallback(spark.comet.exec.strictFloatingPoint=true) -SELECT id, array_max(d), array_max(f) FROM test_array_extrema_strict_fallback - -query expect_fallback(spark.comet.exec.strictFloatingPoint=true) -SELECT id, array_min(array(array(d[0]), array(d[1]))), - array_max(array(array(f[0]), array(f[1]))) -FROM test_array_extrema_strict_fallback WHERE id IN (1, 2) - -query -SELECT id, array_min(array(a, b)), array_max(array(a, b)) FROM test_array_extrema_strict_fallback diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_opt_in.sql b/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_opt_in.sql deleted file mode 100644 index 4617ff2a353..00000000000 --- a/spark/src/test/resources/sql-tests/expressions/array/array_extrema_strict_fp_opt_in.sql +++ /dev/null @@ -1,42 +0,0 @@ --- 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. - --- Explicitly allowing incompatible extrema must retain native execution in strict mode, --- even with the codegen dispatcher disabled. These arrays have no signed-zero ties, whose --- native parity remains tracked by https://github.com/apache/datafusion-comet/issues/5401. - --- Config: spark.comet.exec.strictFloatingPoint=true --- Config: spark.comet.exec.scalaUDF.codegen.enabled=false --- Config: spark.comet.expression.ArrayMin.allowIncompatible=true --- Config: spark.comet.expression.ArrayMax.allowIncompatible=true --- ConfigMatrix: parquet.enable.dictionary=false,true - -statement -CREATE TABLE test_array_extrema_strict_opt_in(id int, d array, f array) USING parquet - -statement -INSERT INTO test_array_extrema_strict_opt_in VALUES - (1, array(double('-3.0'), double('1.0'), double('2.0')), array(float('-3.0'), float('1.0'), float('2.0'))), - (2, array(double('0.0')), array(float('0.0'))), - (3, array(double('-0.0')), array(float('-0.0'))), - (4, array(NULL, double('-2.0'), double('4.0')), array(NULL, float('-2.0'), float('4.0'))), - (5, array(), array()), - (6, NULL, NULL) - -query -SELECT id, array_min(d), array_max(d), array_min(f), array_max(f) -FROM test_array_extrema_strict_opt_in 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 a84e41314f3..9fd43065605 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 @@ -69,8 +72,8 @@ INSERT INTO test_array_max_float VALUES query SELECT array_max(arr) FROM test_array_max_float --- Spark preserves the first equal zero (-0.0 here); the native maximum returns +0.0. --- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. +-- 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 @@ -78,5 +81,34 @@ statement INSERT INTO test_array_max_negzero VALUES (array(double('-0.0'), double('0.0')), array(float('-0.0'), float('0.0'))) -query ignore(https://github.com/apache/datafusion-comet/issues/5401) +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 00000000000..8451138c578 --- /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 00000000000..de6e7d34efd --- /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_max_strict_fp.sql b/spark/src/test/resources/sql-tests/expressions/array/array_max_strict_fp.sql deleted file mode 100644 index 281894dba2e..00000000000 --- a/spark/src/test/resources/sql-tests/expressions/array/array_max_strict_fp.sql +++ /dev/null @@ -1,55 +0,0 @@ --- 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. - --- Strict mode uses Spark's codegen for signed-zero ties. Spark retains the first equal --- extremum, so both input orders and floating-point widths must be checked without a tolerance. --- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. - --- Config: spark.comet.exec.strictFloatingPoint=true --- Config: spark.comet.exec.scalaUDF.codegen.enabled=true --- Config: spark.comet.expression.ArrayMax.allowIncompatible=false --- ConfigMatrix: parquet.enable.dictionary=false,true - -statement -CREATE TABLE test_array_max_strict_fp(id int, d array, f array) USING parquet - -statement -INSERT INTO test_array_max_strict_fp VALUES - (1, array(double('0.0'), double('-0.0')), array(float('0.0'), float('-0.0'))), - (2, array(double('-0.0'), double('0.0')), array(float('-0.0'), float('0.0'))), - (3, array(NULL, double('0.0'), double('-0.0')), array(NULL, float('0.0'), float('-0.0'))), - (4, array(NULL, double('-0.0'), double('0.0')), array(NULL, float('-0.0'), float('0.0'))), - (5, array(double('-0.0')), array(float('-0.0'))), - (6, array(), array()), - (7, array(NULL), array(NULL)), - (8, NULL, NULL) - -query -SELECT id, array_max(d), array_max(f) FROM test_array_max_strict_fp - --- The result type contains a float/double inside an array, so the strict-mode guard is recursive. -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_strict_fp WHERE id IN (1, 2) - --- The SQL harness disables constant folding, so literal inputs exercise the dispatcher 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'))) 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 fe825675f42..b2d1f44e416 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 preserves the first equal zero (+0.0 here); the native minimum returns -0.0. --- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. +-- 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(https://github.com/apache/datafusion-comet/issues/5401) +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 preserves the first equal zero (+0.0 here); the native minimum returns -0.0. --- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. +-- 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(https://github.com/apache/datafusion-comet/issues/5401) +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 00000000000..42c7678e283 --- /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 00000000000..0002b8f4b6f --- /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/resources/sql-tests/expressions/array/array_min_strict_fp.sql b/spark/src/test/resources/sql-tests/expressions/array/array_min_strict_fp.sql deleted file mode 100644 index f0106b0e3d0..00000000000 --- a/spark/src/test/resources/sql-tests/expressions/array/array_min_strict_fp.sql +++ /dev/null @@ -1,55 +0,0 @@ --- 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. - --- Strict mode uses Spark's codegen for signed-zero ties. Spark retains the first equal --- extremum, so both input orders and floating-point widths must be checked without a tolerance. --- Native parity is tracked by https://github.com/apache/datafusion-comet/issues/5401. - --- Config: spark.comet.exec.strictFloatingPoint=true --- Config: spark.comet.exec.scalaUDF.codegen.enabled=true --- Config: spark.comet.expression.ArrayMin.allowIncompatible=false --- ConfigMatrix: parquet.enable.dictionary=false,true - -statement -CREATE TABLE test_array_min_strict_fp(id int, d array, f array) USING parquet - -statement -INSERT INTO test_array_min_strict_fp VALUES - (1, array(double('0.0'), double('-0.0')), array(float('0.0'), float('-0.0'))), - (2, array(double('-0.0'), double('0.0')), array(float('-0.0'), float('0.0'))), - (3, array(NULL, double('0.0'), double('-0.0')), array(NULL, float('0.0'), float('-0.0'))), - (4, array(NULL, double('-0.0'), double('0.0')), array(NULL, float('-0.0'), float('0.0'))), - (5, array(double('-0.0')), array(float('-0.0'))), - (6, array(), array()), - (7, array(NULL), array(NULL)), - (8, NULL, NULL) - -query -SELECT id, array_min(d), array_min(f) FROM test_array_min_strict_fp - --- The result type contains a float/double inside an array, so the strict-mode guard is recursive. -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_strict_fp WHERE id IN (1, 2) - --- The SQL harness disables constant folding, so literal inputs exercise the dispatcher 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'))) diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index 05a6e8e650d..4f4637fccdd 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) From 5dbbc3db062bfb8e308a800d33a44e174f18dcce Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Sat, 22 Aug 2026 15:11:39 -0700 Subject: [PATCH 3/3] fix: bound native array extrema result buffers --- .../src/array_funcs/array_extrema.rs | 75 +++++++++- .../src/array_funcs/array_extrema/tests.rs | 129 ++++++++++++++++++ 2 files changed, 201 insertions(+), 3 deletions(-) diff --git a/native/spark-expr/src/array_funcs/array_extrema.rs b/native/spark-expr/src/array_funcs/array_extrema.rs index d9858ac165e..854cc337d7f 100644 --- a/native/spark-expr/src/array_funcs/array_extrema.rs +++ b/native/spark-expr/src/array_funcs/array_extrema.rs @@ -19,8 +19,9 @@ use std::cmp::Ordering; use std::sync::Arc; use arrow::array::{ - make_comparator, new_empty_array, Array, ArrayRef, AsArray, DynComparator, GenericListArray, - GenericListViewArray, OffsetSizeTrait, PrimitiveArray, PrimitiveBuilder, UInt64Array, + 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}; @@ -207,7 +208,75 @@ fn nested_extrema( } // Take from the original values, not comparator-normalized or reconstructed values. // This preserves nested fields, dictionary types, signed zeros, and NaN payloads. - Ok(take(values.as_ref(), &UInt64Array::from(indices), None)?) + 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: diff --git a/native/spark-expr/src/array_funcs/array_extrema/tests.rs b/native/spark-expr/src/array_funcs/array_extrema/tests.rs index 9521d317c04..bd532774d84 100644 --- a/native/spark-expr/src/array_funcs/array_extrema/tests.rs +++ b/native/spark-expr/src/array_funcs/array_extrema/tests.rs @@ -200,6 +200,135 @@ fn assert_winners( } } +#[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 {