You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/03 13:16:33 UTC
(datafusion-comet) 01/01: fix: gate array distinct and union signed-zero semantics by Spark version (#5750)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5750-4b8334b9adc35a515964185e4648f7769b6cb083
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 71250f2bc725e8b101650195a0a68acfbc559ddf
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Sat Oct 3 12:57:34 2026 +0000
fix: gate array distinct and union signed-zero semantics by Spark version (#5750)
* fix: gate array distinct and union signed-zero semantics by Spark version
* test: preserve native array distinct fuzz coverage on older Spark
* test: separate signed-zero array fixtures by Spark version
* docs: explain array set fallback tradeoffs and tighten coverage
* Clarify array set NaN compatibility and strengthen regression coverage
* test: check the version-gated signed-zero array fixtures
The signed-zero array cases moved out of array_distinct/except/intersect/union.sql
into version-gated fixtures, so the #5702 parser invariant failed on its old layout.
Check instead that the four general fixtures carry no signed-zero cases, that each
gated fixture keeps a signed-zero literal query for array_distinct and array_union,
and that only the Spark 4.2+ fixture asserts native agreement: Spark 3.x expects
the fallback and 4.0/4.1 compare answers only.
* test: drop the #5701 float semantics known gap
array_distinct and array_union now fall back on Spark versions without
SPARK-54918, so the sweep's cases for them match Spark and the suite asks
for the entry to be removed. Remove it with its version helper, which
nothing else uses.
* fix: run floating-point array_distinct and array_union natively only on Spark 4.2.0
The native kernels match Spark only when Spark's optimizer has normalized the
arguments in the plan, and 4.2.0 is the only release with that rewrite
(SPARK-54918). SPARK-59602 replaced it in 4.0.5, 4.1.4 and 4.2.1 with
normalization during evaluation, which the native kernels do not reproduce
for NaN payloads or nested zeros. Report Incompatible for floating-point
elements on every other version, so they fall back unless opted in.
Pin the boundary over version strings, add a nested signed-zero test next to
the NaN test, assert the fallback in the 4.0/4.1 fixture, opt the 4.2+ fixture
into the native path so it holds on every 4.2 patch release, and update the
floating-point guide and expression table.
* docs: correct the array_distinct and array_union float notes in the array audit
The audit said Spark's SQLOpenHashSet canonicalizes +0.0/-0.0 and that native
results match Spark for NaN and signed zero. Spark before SPARK-54918 keeps the
zeros apart, and native execution now runs only on Spark 4.2.0 for float
element types, so point at the floating-point compatibility section instead.
---
.../expression-audits/array_funcs.md | 6 +-
.../latest/compatibility/floating-point.md | 35 ++++++
docs/source/user-guide/latest/expressions.md | 4 +-
.../org/apache/comet/serde/QueryPlanSerde.scala | 2 +-
.../main/scala/org/apache/comet/serde/arrays.scala | 42 ++++++-
.../org/apache/spark/sql/comet/util/Utils.scala | 3 +
.../sql-tests/expressions/array/array_distinct.sql | 42 +------
.../sql-tests/expressions/array/array_except.sql | 35 ------
.../expressions/array/array_intersect.sql | 70 +-----------
.../expressions/array/array_set_signed_zero.sql | 96 ++++++++++++++++
.../array/array_set_signed_zero_spark_3.sql | 89 +++++++++++++++
.../array/array_set_signed_zero_spark_4_0_4_1.sql | 92 +++++++++++++++
.../sql-tests/expressions/array/array_union.sql | 34 +-----
.../apache/comet/CometArrayExpressionSuite.scala | 126 ++++++++++++++++++++-
.../org/apache/comet/CometCodegenFuzzSuite.scala | 27 +++--
.../apache/comet/CometFloatSemanticsSuite.scala | 18 ---
.../org/apache/comet/SqlFileTestParserSuite.scala | 87 ++++++++++----
17 files changed, 572 insertions(+), 236 deletions(-)
diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md
index 0911c45aff..2eed573d7f 100644
--- a/docs/source/contributor-guide/expression-audits/array_funcs.md
+++ b/docs/source/contributor-guide/expression-audits/array_funcs.md
@@ -53,10 +53,10 @@
## array_distinct
- Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayDistinct(child)` over `ArraySetLike`; uses `SQLOpenHashSet` so NaN and `+0.0`/`-0.0` are canonicalized. Wired as `CometScalarFunction("array_distinct")`.
+- Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayDistinct(child)` over `ArraySetLike`; uses `SQLOpenHashSet`, which merges NaNs but keeps `+0.0` and `-0.0` distinct. Wired as `CometScalarFunction("array_distinct")`.
- 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 and signed zero match Spark; DataFusion canonicalizes them like Spark's `SQLOpenHashSet`.
+- Float/double element types run natively only on Spark 4.2.0, which normalizes the arguments in the plan (SPARK-54918); every other version falls back by default. See [Array distinct and union](../../user-guide/latest/compatibility/floating-point.md#array-distinct-and-union).
## array_except
@@ -141,7 +141,7 @@
- Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayUnion(left, right) extends ArrayBinaryLike with ComplexTypeMergingExpression`; result is left-side distinct elements followed by new right-side elements. Wired as `CometScalarFunction("array_union")`.
- Spark 4.0.1 (audited 2026-05-27): `nullIntolerant = true` moves into `ArrayBinaryLike`; overflow path uses `arrayFunctionWithElementsExceedLimitError`.
- Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
-- Float/double NaN and signed-zero canonicalization matches `array_distinct`. Result element ordering also matches Spark (left-side distinct elements followed by new right-side elements).
+- Float/double element types follow the same Spark 4.2.0-only gate as `array_distinct`. Result element ordering matches Spark (left-side distinct elements followed by new right-side elements).
## arrays_overlap
diff --git a/docs/source/user-guide/latest/compatibility/floating-point.md b/docs/source/user-guide/latest/compatibility/floating-point.md
index 0ffd7313ac..ab17d16309 100644
--- a/docs/source/user-guide/latest/compatibility/floating-point.md
+++ b/docs/source/user-guide/latest/compatibility/floating-point.md
@@ -75,6 +75,41 @@ greater than non-NaN values. The original first equal element is retained: for e
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.
+## Array distinct and union
+
+`array_distinct` and `array_union` fall back to Spark when their element type contains
+`FLOAT` or `DOUBLE`, on every Spark version except 4.2.0. Spark 4.2.0 normalizes signed zeros
+and NaNs in the arguments of these functions before they run (SPARK-54918), so native execution
+returns the same results. Spark 3.4, 3.5, 4.0.0 to 4.0.4, and 4.1.0 to 4.1.3 keep positive and
+negative zero distinct in flat arrays. Spark 4.0.5+, 4.1.4+, and 4.2.1+ normalize while these
+functions evaluate instead (SPARK-59602), which native execution does not match for NaNs or for
+zeros nested in arrays or structs. Other element types remain native.
+
+The check is based on the element type, not the values. It also applies to NULL or empty
+floating-point arrays and columns that never contain negative zero. The entire projection
+falls back to Spark, introducing a `CometColumnarToRow` transition and moving unrelated
+expressions in the same projection out of Comet. For example,
+`SELECT id + 1, array_distinct(a), i[0] + 5` evaluates all three expressions in a Spark `Project`.
+
+This can have a substantial cost. A local Spark 4.1.3 benchmark of
+`sum(cardinality(array_distinct(d)))` over two million `array<double>` rows found the default
+projection fallback about 15 times slower than native opt-in (best of five runs).
+The slowdown depends on the workload.
+
+Setting `spark.comet.expression.ArrayDistinct.allowIncompatible=true` or
+`spark.comet.expression.ArrayUnion.allowIncompatible=true` restores native execution on other
+versions, but signed-zero and NaN results may differ from Spark. Native execution can keep
+NaNs with different signs or payloads distinct. Signed-zero differences also depend on the
+element type: native execution merges positive and negative zero in flat floating-point arrays,
+but can keep them distinct inside nested arrays or structs. Only opt in if these differences
+are acceptable for your data.
+
+The check uses the Spark version number, not the changes a build contains. A build that reports
+any version other than 4.2.0 falls back even if it includes SPARK-54918. A vendor build that
+reports 4.2.0 but includes SPARK-59602 still runs natively; set
+`spark.comet.expression.ArrayDistinct.enabled=false` and
+`spark.comet.expression.ArrayUnion.enabled=false` on such a build.
+
## `array_remove` and `sort_array`
`array_remove` compares `FLOAT` and `DOUBLE` elements as Spark does: `-0.0` equals `0.0`, and all
diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md
index 3c1e868432..7cf2808565 100644
--- a/docs/source/user-guide/latest/expressions.md
+++ b/docs/source/user-guide/latest/expressions.md
@@ -156,7 +156,7 @@ The tables below list every Spark built-in expression with its current status.
| `array_append` | ✅ | Native | |
| `array_compact` | ✅ | — | |
| `array_contains` | ✅ | Native | Float/double element arrays route through the JVM codegen dispatcher by default; the native path is opt-in via allowIncompatible |
-| `array_distinct` | ✅ | Native | NaN/signed-zero handling may differ ([details](compatibility/floating-point.md)) |
+| `array_distinct` | ✅ | Native | Floating-point elements fall back on Spark versions other than 4.2.0; signed-zero and NaN results may differ with native opt-in ([details](compatibility/floating-point.md)) |
| `array_except` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the incompatible native path is opt-in via allowIncompatible ([details](compatibility/expressions/array.md)) |
| `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)) |
@@ -167,7 +167,7 @@ The tables below list every Spark built-in expression with its current status.
| `array_prepend` | ✅ | — | |
| `array_remove` | ✅ | Native | |
| `array_repeat` | ✅ | Native | |
-| `array_union` | ✅ | Native | NaN/signed-zero handling may differ ([details](compatibility/floating-point.md)) |
+| `array_union` | ✅ | Native | Floating-point elements fall back on Spark versions other than 4.2.0; signed-zero and NaN results may differ with native opt-in ([details](compatibility/floating-point.md)) |
| `arrays_overlap` | ✅ | Native | |
| `arrays_zip` | ✅ | Native | |
| `element_at` | ✅ | Native | |
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index 98a33acc99..bcdb37122b 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -63,7 +63,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
// through CometArrayFilter -> CometArrayCompact -> DataFusion's array_compact. On Spark
// 4.0+ the rewrite is wrapped in KnownNotContainsNull, stripped by Spark4xCometExprShim.
classOf[ArrayContains] -> CometArrayContains,
- classOf[ArrayDistinct] -> CometScalarFunction("array_distinct"),
+ classOf[ArrayDistinct] -> CometArrayDistinct,
classOf[ArrayExcept] -> CometArrayExcept,
classOf[ArrayFilter] -> CometArrayFilter,
classOf[ArrayInsert] -> CometArrayInsert,
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 6a5ce01561..b2a75f8478 100644
--- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
@@ -22,8 +22,10 @@ package org.apache.comet.serde
import scala.annotation.tailrec
import scala.jdk.CollectionConverters._
-import org.apache.spark.sql.catalyst.expressions.{And, ArrayAggregate, ArrayAppend, ArrayContains, ArrayExcept, ArrayExists, ArrayFilter, ArrayForAll, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayPosition, ArrayRemove, ArraySort, ArraysOverlap, ArraysZip, ArrayTransform, ArrayUnion, Attribute, BoundReference, Cast, CreateArray, ElementAt, EmptyRow, Expression, Flatten, GetArrayItem, IsNotNull, IsNull, LambdaFunction, Literal, NamedLambdaVariable, Reverse, Sequence, Si [...]
+import org.apache.spark.SPARK_VERSION
+import org.apache.spark.sql.catalyst.expressions.{And, ArrayAggregate, ArrayAppend, ArrayContains, ArrayDistinct, ArrayExcept, ArrayExists, ArrayFilter, ArrayForAll, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayPosition, ArrayRemove, ArraySort, ArraysOverlap, ArraysZip, ArrayTransform, ArrayUnion, Attribute, BoundReference, Cast, CreateArray, ElementAt, EmptyRow, Expression, Flatten, GetArrayItem, IsNotNull, IsNull, LambdaFunction, Literal, NamedLambdaVariable, Revers [...]
import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.comet.util.Utils
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -536,7 +538,45 @@ object CometSlice extends CometExpressionSerde[Slice] {
}
}
+private[comet] object ArraySetSupport {
+ val floatingPointReason: String =
+ "Floating-point elements match Spark's signed-zero and NaN semantics natively only on " +
+ "Spark 4.2.0, whose optimizer normalizes the arguments (SPARK-54918)"
+
+ // The native kernels match Spark only when the plan has already normalized the arguments, and
+ // only Spark 4.2.0 does that (SPARK-54918). Earlier releases keep flat signed zeros apart.
+ // From 4.0.5, 4.1.4 and 4.2.1, SPARK-59602 normalizes during evaluation instead, which the
+ // native kernels do not match for NaN payloads or nested zeros. A top-level
+ // KnownFloatingPointNormalized marker cannot replace the version check: Spark also normalizes
+ // CreateArray, If, CaseWhen, and Coalesce recursively without wrapping the resulting array.
+ def normalizesArgumentsInPlan(version: String): Boolean =
+ Utils.majorMinorPatchVersion(version).contains((4, 2, 0))
+
+ def supportLevel(dataType: DataType): SupportLevel = {
+ if (SupportLevel.containsType(dataType, classOf[FloatType], classOf[DoubleType]) &&
+ !normalizesArgumentsInPlan(SPARK_VERSION)) {
+ Incompatible(Some(floatingPointReason))
+ } else {
+ Compatible()
+ }
+ }
+}
+
+// Use projection fallback to avoid codegen dispatch overhead for array-valued results.
+// The native implementation remains available through opt-in.
+object CometArrayDistinct extends CometScalarFunction[ArrayDistinct]("array_distinct") {
+ override def getIncompatibleReasons(): Seq[String] = Seq(ArraySetSupport.floatingPointReason)
+
+ override def getSupportLevel(expr: ArrayDistinct): SupportLevel =
+ ArraySetSupport.supportLevel(expr.dataType)
+}
+
object CometArrayUnion extends CometExpressionSerde[ArrayUnion] {
+ override def getIncompatibleReasons(): Seq[String] = Seq(ArraySetSupport.floatingPointReason)
+
+ override def getSupportLevel(expr: ArrayUnion): SupportLevel =
+ ArraySetSupport.supportLevel(expr.dataType)
+
override def convert(
expr: ArrayUnion,
inputs: Seq[Attribute],
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala b/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala
index eb5947f30c..c884344ef1 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala
@@ -51,6 +51,9 @@ import org.apache.comet.vector.CometVector
object Utils extends CometTypeShim with Logging {
private val VariantExtensionName = "arrow.parquet.variant"
+ def majorMinorPatchVersion(version: String): Option[(Int, Int, Int)] =
+ org.apache.spark.util.VersionUtils.majorMinorPatchVersion(version)
+
def getConfPath(confFileName: String): String = {
sys.env
.get(COMET_CONF_DIR_ENV)
diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_distinct.sql b/spark/src/test/resources/sql-tests/expressions/array/array_distinct.sql
index a0ca51a11a..0a6ff4b138 100644
--- a/spark/src/test/resources/sql-tests/expressions/array/array_distinct.sql
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_distinct.sql
@@ -15,6 +15,9 @@
-- specific language governing permissions and limitations
-- under the License.
+-- No negative zeros in this fixture; signed-zero compatibility is tested in array_set_signed_zero*.
+-- Config: spark.comet.expression.ArrayDistinct.allowIncompatible=true
+
-- ===== INT arrays =====
statement
@@ -136,29 +139,6 @@ SELECT array_distinct(array(CAST('NaN' AS DOUBLE), NULL, CAST('NaN' AS DOUBLE),
query
SELECT array_distinct(array(CAST('Infinity' AS DOUBLE), CAST('-Infinity' AS DOUBLE), CAST('Infinity' AS DOUBLE), 0.0))
--- negative zero (literal). NormalizeFloatingNumbers only rewrites grouping keys,
--- join keys, window partition specs and Distinct; a plain SELECT keeps the -0.0
--- literal intact. Spark therefore keeps 0.0 and -0.0 distinct while Comet
--- (DataFusion) collapses them: array_distinct([0.0, -0.0, 1.0]) is [0.0, -0.0, 1.0]
--- in Spark but [0.0, 1.0] in Comet. Skip until Spark normalizes these zeros
--- (Spark 4.2+, SPARK-54918).
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT array_distinct(array(0.0, double('-0.0'), 1.0))
-
--- negative zero (column-sourced). Same divergence as the literal case above:
--- NormalizeFloatingNumbers does not rewrite array-function inputs, so Spark keeps
--- -0.0 distinct from 0.0 while Comet collapses them. Skip until Spark normalizes
--- these zeros (Spark 4.2+, SPARK-54918).
-statement
-CREATE TABLE test_array_distinct_dbl_negzero(arr array<double>) USING parquet
-
-statement
-INSERT INTO test_array_distinct_dbl_negzero VALUES
- (array(0.0, double('-0.0'), 1.0))
-
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT array_distinct(arr) FROM test_array_distinct_dbl_negzero
-
-- ===== FLOAT arrays =====
statement
@@ -181,22 +161,6 @@ SELECT array_distinct(arr) FROM test_array_distinct_float
query
SELECT array_distinct(array(CAST('NaN' AS FLOAT), CAST('NaN' AS FLOAT), CAST(1.0 AS FLOAT)))
--- negative zero (literal). Same divergence as the double literal case above.
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT array_distinct(array(CAST(0.0 AS FLOAT), float('-0.0'), CAST(1.0 AS FLOAT)))
-
--- negative zero (column-sourced). Same divergence as the double cases above, skipped
--- until Spark normalizes these zeros (Spark 4.2+, SPARK-54918).
-statement
-CREATE TABLE test_array_distinct_flt_negzero(arr array<float>) USING parquet
-
-statement
-INSERT INTO test_array_distinct_flt_negzero VALUES
- (array(CAST(0.0 AS FLOAT), float('-0.0'), CAST(1.0 AS FLOAT)))
-
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT array_distinct(arr) FROM test_array_distinct_flt_negzero
-
-- ===== DECIMAL arrays =====
statement
diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_except.sql b/spark/src/test/resources/sql-tests/expressions/array/array_except.sql
index 47504b25ad..6ec4c275d1 100644
--- a/spark/src/test/resources/sql-tests/expressions/array/array_except.sql
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_except.sql
@@ -71,38 +71,3 @@ INSERT INTO test_except_float VALUES
query
SELECT a, b, array_except(a, b) FROM test_except_float
-
--- negative zero (literal). Same signed-zero divergence as the column-sourced
--- cases below: NormalizeFloatingNumbers does not rewrite array-function inputs,
--- so a plain SELECT keeps the -0.0 literal intact. Spark keeps -0.0 distinct
--- from 0.0 while Comet (DataFusion) collapses them. Skip until Spark normalizes
--- these zeros (Spark 4.2+, SPARK-54918).
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT array_except(array(0.0, double('-0.0'), 1.0), array(0.0)),
- array_except(array(0.0, 1.0), array(double('-0.0')))
-
--- negative zero (column-sourced). Spark keeps -0.0 distinct from 0.0 while Comet
--- (DataFusion) collapses them, so array_except([0.0, -0.0, 1.0], [0.0]) is [-0.0, 1.0]
--- in Spark but [1.0] in Comet. The divergence is not limited to parquet columns;
--- the literal case above is skipped for the same reason. Skip until Spark
--- normalizes these zeros (Spark 4.2+, SPARK-54918).
-statement
-CREATE TABLE test_except_dbl_negzero(a array<double>, b array<double>) USING parquet
-
-statement
-INSERT INTO test_except_dbl_negzero VALUES
- (array(0.0, double('-0.0'), 1.0), array(0.0)),
- (array(0.0, 1.0), array(double('-0.0')))
-
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT a, b, array_except(a, b) FROM test_except_dbl_negzero
-
-statement
-CREATE TABLE test_except_flt_negzero(a array<float>, b array<float>) USING parquet
-
-statement
-INSERT INTO test_except_flt_negzero VALUES
- (array(cast(0.0 as float), float('-0.0'), cast(1.0 as float)), array(cast(0.0 as float)))
-
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT a, b, array_except(a, b) FROM test_except_flt_negzero
diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_intersect.sql b/spark/src/test/resources/sql-tests/expressions/array/array_intersect.sql
index e7a5f0a775..6279c104f3 100644
--- a/spark/src/test/resources/sql-tests/expressions/array/array_intersect.sql
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_intersect.sql
@@ -16,7 +16,6 @@
-- under the License.
-- Config: spark.comet.expression.ArrayIntersect.allowIncompatible=true
--- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
-- DataFusion's array_intersect emits elements in the order of the longer input
-- (it uses the shorter side as the hash lookup), while Spark emits elements in
@@ -31,7 +30,7 @@ CREATE TABLE test_array_intersect(a array<int>, b array<int>) USING parquet
statement
INSERT INTO test_array_intersect VALUES (array(1, 2, 3), array(2, 3, 4)), (array(1, 2), array(3, 4)), (array(), array(1)), (NULL, array(1)), (array(1, NULL), array(NULL, 2))
-query expect_native(array_intersect)
+query
SELECT array_intersect(a, b) FROM test_array_intersect
-- column + literal
@@ -122,7 +121,7 @@ query
SELECT a, b, array_intersect(a, b) FROM test_intersect_long
-- Float arrays with NaN, Infinity, and -Infinity. Signed-zero membership cases are
--- isolated below (see the SPARK-54918 note).
+-- covered in array_set_signed_zero*.sql.
statement
CREATE TABLE test_intersect_float(a array<float>, b array<float>) USING parquet
@@ -139,7 +138,7 @@ INSERT INTO test_intersect_float VALUES
query
SELECT a, b, array_intersect(a, b) FROM test_intersect_float
--- Double arrays with NaN, Infinity, and -Infinity. Signed-zero cases isolated below.
+-- Double arrays with NaN, Infinity, and -Infinity. Signed-zero cases are in array_set_signed_zero*.sql.
statement
CREATE TABLE test_intersect_dbl(a array<double>, b array<double>) USING parquet
@@ -157,46 +156,6 @@ INSERT INTO test_intersect_dbl VALUES
query
SELECT a, b, array_intersect(a, b) FROM test_intersect_dbl
--- negative zero (literal). Same signed-zero membership divergence as the
--- column-sourced cases below: NormalizeFloatingNumbers does not rewrite
--- array-function inputs, so a plain SELECT keeps the -0.0 literal intact.
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT array_intersect(array(double('-0.0')), array(double('0.0'))),
- array_intersect(array(double('0.0')), array(double('-0.0'))),
- array_intersect(array(double('-0.0')), array(double('-0.0')))
-
--- Signed-zero membership. Spark keeps -0.0 distinct from 0.0 while Comet (DataFusion)
--- collapses them, so array_intersect([-0.0], [0.0]) is [] in Spark but [0.0] in Comet,
--- and array_intersect([-0.0], [-0.0]) is [-0.0] in Spark but [0.0] in Comet.
--- The divergence is not limited to parquet columns; the literal case above is
--- skipped for the same reason. Skip until Spark normalizes these zeros
--- (Spark 4.2+, SPARK-54918).
-statement
-CREATE TABLE test_intersect_flt_negzero(a array<float>, b array<float>) USING parquet
-
-statement
-INSERT INTO test_intersect_flt_negzero VALUES
- (array(cast(0.0 as float), float('-0.0')), array(cast(0.0 as float))),
- (array(float('-0.0')), array(float('0.0'))),
- (array(float('0.0')), array(float('-0.0'))),
- (array(float('-0.0')), array(float('-0.0')))
-
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT a, b, array_intersect(a, b) FROM test_intersect_flt_negzero
-
-statement
-CREATE TABLE test_intersect_dbl_negzero(a array<double>, b array<double>) USING parquet
-
-statement
-INSERT INTO test_intersect_dbl_negzero VALUES
- (array(0.0, double('-0.0')), array(0.0)),
- (array(double('-0.0')), array(double('0.0'))),
- (array(double('0.0')), array(double('-0.0'))),
- (array(double('-0.0')), array(double('-0.0')))
-
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT a, b, array_intersect(a, b) FROM test_intersect_dbl_negzero
-
-- decimal arrays
statement
CREATE TABLE test_intersect_dec(a array<decimal(10,2)>, b array<decimal(10,2)>) USING parquet
@@ -267,26 +226,3 @@ SELECT array_intersect(array(1, NULL, 3), b) FROM test_array_intersect
-- conditional (CASE WHEN) arrays
query
SELECT array_intersect(CASE WHEN a IS NOT NULL THEN a ELSE array(0) END, b) FROM test_array_intersect
-
--- Without native opt-in, preserve Spark's left-input order even when the right array is longer.
-statement
-INSERT INTO test_array_intersect VALUES (array(2, 1), array(3, 1, 2))
-
-statement
-SET spark.comet.expression.ArrayIntersect.allowIncompatible=false
-
-query expect_dispatch(array_intersect)
-SELECT array_intersect(a, b) FROM test_array_intersect
-
-statement
-SET spark.comet.exec.scalaUDF.codegen.enabled=false
-
-query expect_fallback(array_intersect: spark.comet.exec.scalaUDF.codegen.enabled=false)
-SELECT array_intersect(a, b) FROM test_array_intersect
-
--- Native opt-in works even with the dispatcher disabled. Normalize the permitted order difference.
-statement
-SET spark.comet.expression.ArrayIntersect.allowIncompatible=true
-
-query expect_native(array_intersect)
-SELECT sort_array(array_intersect(a, b)) FROM test_array_intersect
diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_set_signed_zero.sql b/spark/src/test/resources/sql-tests/expressions/array/array_set_signed_zero.sql
new file mode 100644
index 0000000000..bc506b5f7d
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_set_signed_zero.sql
@@ -0,0 +1,96 @@
+-- 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.2
+
+-- Every 4.2+ release normalizes signed zeros in these functions: 4.2.0 in the plan (SPARK-54918),
+-- later releases during evaluation (SPARK-59602). The native kernels normalize flat zeros too, so
+-- require native execution through the opt-ins. By default distinct/union run natively only on
+-- 4.2.0; CometArrayExpressionSuite covers that choice and the NaN and nested cases.
+-- Config: spark.comet.expression.ArrayDistinct.allowIncompatible=true
+-- Config: spark.comet.expression.ArrayUnion.allowIncompatible=true
+-- Config: spark.comet.expression.ArrayExcept.allowIncompatible=true
+-- Config: spark.comet.expression.ArrayIntersect.allowIncompatible=true
+
+statement
+CREATE TABLE test_array_set_signed_zero_float(a array<float>, b array<float>) USING parquet
+
+statement
+INSERT INTO test_array_set_signed_zero_float VALUES
+ (array(float('0.0'), float('-0.0'), float('1.0')), array(float('0.0'))),
+ (array(float('0.0'), float('1.0')), array(float('-0.0'))),
+ (array(float('-0.0')), array(float('0.0'))),
+ (array(float('0.0')), array(float('-0.0'))),
+ (array(float('-0.0')), array(float('-0.0'))),
+ (array(float('0.0'), float('-0.0')), array(float('0.0')))
+
+query
+SELECT array_distinct(array(float('0.0'), float('-0.0'), float('1.0')))
+
+query
+SELECT array_union(array(float('0.0')), array(float('-0.0')))
+
+query
+SELECT a, b, array_distinct(a) FROM test_array_set_signed_zero_float
+
+query
+SELECT a, b, array_union(a, b) FROM test_array_set_signed_zero_float
+
+query
+SELECT a, b, array_except(a, b) FROM test_array_set_signed_zero_float
+
+query
+SELECT a, b, array_intersect(a, b) FROM test_array_set_signed_zero_float
+
+statement
+CREATE TABLE test_array_set_signed_zero_double(a array<double>, b array<double>) USING parquet
+
+statement
+INSERT INTO test_array_set_signed_zero_double VALUES
+ (array(double('0.0'), double('-0.0'), double('1.0')), array(double('0.0'))),
+ (array(double('0.0'), double('1.0')), array(double('-0.0'))),
+ (array(double('-0.0')), array(double('0.0'))),
+ (array(double('0.0')), array(double('-0.0'))),
+ (array(double('-0.0')), array(double('-0.0'))),
+ (array(double('0.0'), double('-0.0')), array(double('0.0')))
+
+query
+SELECT array_distinct(array(double('0.0'), double('-0.0'), double('1.0')))
+
+query
+SELECT array_union(array(double('0.0')), array(double('-0.0')))
+
+query
+SELECT array_except(array(double('0.0'), double('-0.0'), double('1.0')), array(double('0.0'))),
+ array_except(array(double('0.0'), double('1.0')), array(double('-0.0')))
+
+query
+SELECT array_intersect(array(double('-0.0')), array(double('0.0'))),
+ array_intersect(array(double('0.0')), array(double('-0.0'))),
+ array_intersect(array(double('-0.0')), array(double('-0.0')))
+
+query
+SELECT a, b, array_distinct(a) FROM test_array_set_signed_zero_double
+
+query
+SELECT a, b, array_union(a, b) FROM test_array_set_signed_zero_double
+
+query
+SELECT a, b, array_except(a, b) FROM test_array_set_signed_zero_double
+
+query
+SELECT a, b, array_intersect(a, b) FROM test_array_set_signed_zero_double
diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_set_signed_zero_spark_3.sql b/spark/src/test/resources/sql-tests/expressions/array/array_set_signed_zero_spark_3.sql
new file mode 100644
index 0000000000..e8da8a97f4
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_set_signed_zero_spark_3.sql
@@ -0,0 +1,89 @@
+-- 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.
+
+-- MaxSparkVersion: 3.5
+
+-- Spark 3 keeps positive and negative zero distinct; these cases can be removed with Spark 3 support.
+
+statement
+CREATE TABLE test_array_set_signed_zero_float(a array<float>, b array<float>) USING parquet
+
+statement
+INSERT INTO test_array_set_signed_zero_float VALUES
+ (array(float('0.0'), float('-0.0'), float('1.0')), array(float('0.0'))),
+ (array(float('0.0'), float('1.0')), array(float('-0.0'))),
+ (array(float('-0.0')), array(float('0.0'))),
+ (array(float('0.0')), array(float('-0.0'))),
+ (array(float('-0.0')), array(float('-0.0'))),
+ (array(float('0.0'), float('-0.0')), array(float('0.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT array_distinct(array(float('0.0'), float('-0.0'), float('1.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT array_union(array(float('0.0')), array(float('-0.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT a, b, array_distinct(a) FROM test_array_set_signed_zero_float
+
+query expect_fallback(SPARK-54918)
+SELECT a, b, array_union(a, b) FROM test_array_set_signed_zero_float
+
+query
+SELECT a, b, array_except(a, b) FROM test_array_set_signed_zero_float
+
+query
+SELECT a, b, array_intersect(a, b) FROM test_array_set_signed_zero_float
+
+statement
+CREATE TABLE test_array_set_signed_zero_double(a array<double>, b array<double>) USING parquet
+
+statement
+INSERT INTO test_array_set_signed_zero_double VALUES
+ (array(double('0.0'), double('-0.0'), double('1.0')), array(double('0.0'))),
+ (array(double('0.0'), double('1.0')), array(double('-0.0'))),
+ (array(double('-0.0')), array(double('0.0'))),
+ (array(double('0.0')), array(double('-0.0'))),
+ (array(double('-0.0')), array(double('-0.0'))),
+ (array(double('0.0'), double('-0.0')), array(double('0.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT array_distinct(array(double('0.0'), double('-0.0'), double('1.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT array_union(array(double('0.0')), array(double('-0.0')))
+
+query
+SELECT array_except(array(double('0.0'), double('-0.0'), double('1.0')), array(double('0.0'))),
+ array_except(array(double('0.0'), double('1.0')), array(double('-0.0')))
+
+query
+SELECT array_intersect(array(double('-0.0')), array(double('0.0'))),
+ array_intersect(array(double('0.0')), array(double('-0.0'))),
+ array_intersect(array(double('-0.0')), array(double('-0.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT a, b, array_distinct(a) FROM test_array_set_signed_zero_double
+
+query expect_fallback(SPARK-54918)
+SELECT a, b, array_union(a, b) FROM test_array_set_signed_zero_double
+
+query
+SELECT a, b, array_except(a, b) FROM test_array_set_signed_zero_double
+
+query
+SELECT a, b, array_intersect(a, b) FROM test_array_set_signed_zero_double
diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_set_signed_zero_spark_4_0_4_1.sql b/spark/src/test/resources/sql-tests/expressions/array/array_set_signed_zero_spark_4_0_4_1.sql
new file mode 100644
index 0000000000..560c5c27bf
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_set_signed_zero_spark_4_0_4_1.sql
@@ -0,0 +1,92 @@
+-- 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
+-- MaxSparkVersion: 4.1
+
+-- 4.0.0-4.0.4 and 4.1.0-4.1.3 keep signed zeros distinct. 4.0.5+ and 4.1.4+ normalize during
+-- evaluation (SPARK-59602), which native distinct/union do not match for NaNs or nested zeros, so
+-- distinct/union fall back on every 4.0 and 4.1 release.
+
+statement
+CREATE TABLE test_array_set_signed_zero_float(a array<float>, b array<float>) USING parquet
+
+statement
+INSERT INTO test_array_set_signed_zero_float VALUES
+ (array(float('0.0'), float('-0.0'), float('1.0')), array(float('0.0'))),
+ (array(float('0.0'), float('1.0')), array(float('-0.0'))),
+ (array(float('-0.0')), array(float('0.0'))),
+ (array(float('0.0')), array(float('-0.0'))),
+ (array(float('-0.0')), array(float('-0.0'))),
+ (array(float('0.0'), float('-0.0')), array(float('0.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT array_distinct(array(float('0.0'), float('-0.0'), float('1.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT array_union(array(float('0.0')), array(float('-0.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT a, b, array_distinct(a) FROM test_array_set_signed_zero_float
+
+query expect_fallback(SPARK-54918)
+SELECT a, b, array_union(a, b) FROM test_array_set_signed_zero_float
+
+query
+SELECT a, b, array_except(a, b) FROM test_array_set_signed_zero_float
+
+query
+SELECT a, b, array_intersect(a, b) FROM test_array_set_signed_zero_float
+
+statement
+CREATE TABLE test_array_set_signed_zero_double(a array<double>, b array<double>) USING parquet
+
+statement
+INSERT INTO test_array_set_signed_zero_double VALUES
+ (array(double('0.0'), double('-0.0'), double('1.0')), array(double('0.0'))),
+ (array(double('0.0'), double('1.0')), array(double('-0.0'))),
+ (array(double('-0.0')), array(double('0.0'))),
+ (array(double('0.0')), array(double('-0.0'))),
+ (array(double('-0.0')), array(double('-0.0'))),
+ (array(double('0.0'), double('-0.0')), array(double('0.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT array_distinct(array(double('0.0'), double('-0.0'), double('1.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT array_union(array(double('0.0')), array(double('-0.0')))
+
+query
+SELECT array_except(array(double('0.0'), double('-0.0'), double('1.0')), array(double('0.0'))),
+ array_except(array(double('0.0'), double('1.0')), array(double('-0.0')))
+
+query
+SELECT array_intersect(array(double('-0.0')), array(double('0.0'))),
+ array_intersect(array(double('0.0')), array(double('-0.0'))),
+ array_intersect(array(double('-0.0')), array(double('-0.0')))
+
+query expect_fallback(SPARK-54918)
+SELECT a, b, array_distinct(a) FROM test_array_set_signed_zero_double
+
+query expect_fallback(SPARK-54918)
+SELECT a, b, array_union(a, b) FROM test_array_set_signed_zero_double
+
+query
+SELECT a, b, array_except(a, b) FROM test_array_set_signed_zero_double
+
+query
+SELECT a, b, array_intersect(a, b) FROM test_array_set_signed_zero_double
diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_union.sql b/spark/src/test/resources/sql-tests/expressions/array/array_union.sql
index 33cd636f8a..27f11bc4ab 100644
--- a/spark/src/test/resources/sql-tests/expressions/array/array_union.sql
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_union.sql
@@ -15,6 +15,9 @@
-- specific language governing permissions and limitations
-- under the License.
+-- No negative zeros in this fixture; signed-zero compatibility is tested in array_set_signed_zero*.
+-- Config: spark.comet.expression.ArrayUnion.allowIncompatible=true
+
statement
CREATE TABLE test_array_union(a array<int>, b array<int>) USING parquet
@@ -144,37 +147,6 @@ INSERT INTO test_union_float VALUES (array(cast(1.0 as float), cast(2.0 as float
query
SELECT a, b, array_union(a, b) FROM test_union_float
--- negative zero (literal). Same signed-zero divergence as the column-sourced
--- cases below: NormalizeFloatingNumbers does not rewrite array-function inputs,
--- so a plain SELECT keeps the -0.0 literal intact. Spark keeps -0.0 distinct
--- from 0.0 while Comet (DataFusion) collapses them. Skip until Spark normalizes
--- these zeros (Spark 4.2+, SPARK-54918).
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT array_union(array(0.0), array(double('-0.0')))
-
--- negative zero (column-sourced). Spark keeps -0.0 distinct from 0.0 while Comet
--- (DataFusion) collapses them, so array_union([0.0], [-0.0]) is [0.0, -0.0] in Spark
--- but [0.0] in Comet. The divergence is not limited to parquet columns; the
--- literal case above is skipped for the same reason. Skip until Spark
--- normalizes these zeros (Spark 4.2+, SPARK-54918).
-statement
-CREATE TABLE test_union_dbl_negzero(a array<double>, b array<double>) USING parquet
-
-statement
-INSERT INTO test_union_dbl_negzero VALUES (array(0.0), array(double('-0.0')))
-
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT a, b, array_union(a, b) FROM test_union_dbl_negzero
-
-statement
-CREATE TABLE test_union_flt_negzero(a array<float>, b array<float>) USING parquet
-
-statement
-INSERT INTO test_union_flt_negzero VALUES (array(cast(0.0 as float)), array(float('-0.0')))
-
-query ignore(https://issues.apache.org/jira/browse/SPARK-54918)
-SELECT a, b, array_union(a, b) FROM test_union_flt_negzero
-
-- boolean arrays
query
SELECT array_union(array(true, false), array(false)) FROM test_union_dbl
diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
index 1c4ad3ce15..5aabafe402 100644
--- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala
@@ -23,21 +23,141 @@ 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, ArrayMax, ArrayMin, ArrayRepeat}
+import org.apache.spark.sql.catalyst.expressions.{ArrayAppend, ArrayDistinct, ArrayExcept, ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayRepeat, ArrayUnion}
import org.apache.spark.sql.catalyst.expressions.{ArrayContains, ArrayRemove}
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, CreateArray, ElementAt, Literal, MonotonicallyIncreasingID}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{ArrayType, StringType}
+import org.apache.spark.sql.types.{ArrayType, DoubleType, FloatType, IntegerType, StringType, StructType}
import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus}
import org.apache.comet.DataTypeSupport.isComplexType
-import org.apache.comet.serde.{CometArrayExcept, CometArrayJoin, CometArrayRemove, CometArrayReverse, CometFlatten, Compatible, ExprOuterClass, Incompatible}
+import org.apache.comet.serde.{ArraySetSupport, CometArrayDistinct, CometArrayExcept, CometArrayJoin, CometArrayRemove, CometArrayReverse, CometArrayUnion, CometFlatten, Compatible, ExprOuterClass, Incompatible}
import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions}
class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
+ test("array set signed-zero Spark patch versions") {
+ // Only 4.2.0 normalizes the arguments in the plan. Earlier releases do not normalize them, and
+ // 4.0.5+, 4.1.4+ and 4.2.1+ normalize during evaluation instead (SPARK-59602).
+ Seq(
+ "3.4.3",
+ "3.5.9",
+ "4.0.4",
+ "4.0.5",
+ "4.0.10",
+ "4.1.3",
+ "4.1.4",
+ "4.1.10",
+ "4.2.1",
+ "4.2.1-SNAPSHOT",
+ "4.2.10",
+ "4.3.0",
+ "5.0.0").foreach { version =>
+ assert(!ArraySetSupport.normalizesArgumentsInPlan(version), version)
+ }
+ Seq("4.2.0", "4.2.0-SNAPSHOT").foreach { version =>
+ assert(ArraySetSupport.normalizesArgumentsInPlan(version), version)
+ }
+ }
+
+ test("array set signed-zero support levels") {
+ // This test covers element-type detection; the preceding test pins the version boundaries.
+ val fixed = ArraySetSupport.normalizesArgumentsInPlan(org.apache.spark.SPARK_VERSION)
+ Seq(FloatType, DoubleType, ArrayType(FloatType), new StructType().add("x", DoubleType))
+ .foreach { elementType =>
+ val child = AttributeReference("a", ArrayType(elementType))()
+ val expected =
+ if (fixed) Compatible() else Incompatible(Some(ArraySetSupport.floatingPointReason))
+ assert(CometArrayDistinct.getSupportLevel(ArrayDistinct(child)) == expected)
+ assert(CometArrayUnion.getSupportLevel(ArrayUnion(child, child)) == expected)
+ }
+ Seq(IntegerType, StringType, ArrayType(IntegerType)).foreach { elementType =>
+ val child = AttributeReference("a", ArrayType(elementType))()
+ assert(CometArrayDistinct.getSupportLevel(ArrayDistinct(child)) == Compatible())
+ assert(CometArrayUnion.getSupportLevel(ArrayUnion(child, child)) == Compatible())
+ }
+ }
+
+ test("array set signed-zero fallback and native opt-in") {
+ withSQLConf(
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
+ "org.apache.spark.sql.catalyst.optimizer.ConstantFolding") {
+ Seq("float", "double").foreach { dataType =>
+ Seq(
+ "array_distinct" -> s"array($dataType('0.0'), $dataType('-0.0'))",
+ "array_union" -> s"array($dataType('0.0')), array($dataType('-0.0'))")
+ .foreach { case (function, arguments) =>
+ val query = s"SELECT $function($arguments)"
+ if (ArraySetSupport.normalizesArgumentsInPlan(org.apache.spark.SPARK_VERSION)) {
+ checkSparkAnswerAndOperator(query)
+ } else {
+ checkSparkAnswerAndFallbackReason(query, "SPARK-54918")
+ }
+ }
+ }
+ Seq(classOf[ArrayDistinct], classOf[ArrayUnion]).foreach { exprClass =>
+ withSQLConf(CometConf.getExprAllowIncompatConfigKey(exprClass) -> "true") {
+ val query = if (exprClass == classOf[ArrayDistinct]) {
+ "SELECT array_distinct(array(double('1.0'), double('1.0')))"
+ } else {
+ "SELECT array_union(array(double('1.0')), array(double('1.0')))"
+ }
+ checkSparkAnswerAndOperator(query)
+ }
+ }
+ }
+ }
+
+ test("array set noncanonical NaN normalization") {
+ withTempDir { dir =>
+ withTempView("array_set_nan") {
+ sql("SELECT float('NaN') AS f, double('NaN') AS d").write.parquet(dir + "/data")
+ spark.read.parquet(dir + "/data").createOrReplaceTempView("array_set_nan")
+ // Negate scanned values because Parquet canonicalizes NaNs on write.
+ Seq("f", "d").foreach { column =>
+ Seq(
+ s"array_distinct(array($column, -$column))",
+ s"array_union(array($column), array(-$column))").foreach { expression =>
+ val query = s"SELECT size($expression) FROM array_set_nan"
+ if (ArraySetSupport.normalizesArgumentsInPlan(org.apache.spark.SPARK_VERSION)) {
+ checkSparkAnswerAndOperator(query)
+ } else {
+ checkSparkAnswerAndFallbackReason(query, "SPARK-54918")
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("array set nested signed-zero normalization") {
+ withTempDir { dir =>
+ withTempView("array_set_zero") {
+ sql("SELECT float('0.0') AS f, double('0.0') AS d").write.parquet(dir + "/data")
+ spark.read.parquet(dir + "/data").createOrReplaceTempView("array_set_zero")
+ // Spark deduplicates nested -0.0 and 0.0 on every version, while the native kernels only
+ // normalize flat zeros.
+ Seq("f", "d").foreach { column =>
+ Seq(
+ s"array_distinct(array(array($column), array(-$column)))",
+ s"array_distinct(array(named_struct('x', $column), named_struct('x', -$column)))",
+ s"array_union(array(array($column)), array(array(-$column)))",
+ s"array_union(array(named_struct('x', $column)), array(named_struct('x', -$column)))")
+ .foreach { expression =>
+ val query = s"SELECT $expression FROM array_set_zero"
+ if (ArraySetSupport.normalizesArgumentsInPlan(org.apache.spark.SPARK_VERSION)) {
+ checkSparkAnswerAndOperator(query)
+ } else {
+ checkSparkAnswerAndFallbackReason(query, "SPARK-54918")
+ }
+ }
+ }
+ }
+ }
+ }
+
test("array_remove - integer") {
withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[ArrayRemove]) -> "true") {
Seq(true, false).foreach { dictionaryEnabled =>
diff --git a/spark/src/test/scala/org/apache/comet/CometCodegenFuzzSuite.scala b/spark/src/test/scala/org/apache/comet/CometCodegenFuzzSuite.scala
index 5c40dccaa3..9c0940ad09 100644
--- a/spark/src/test/scala/org/apache/comet/CometCodegenFuzzSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometCodegenFuzzSuite.scala
@@ -28,7 +28,7 @@ import org.apache.commons.io.FileUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.CometTestBase
import org.apache.spark.sql.catalyst.CatalystTypeConverters
-import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, Literal}
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, GenericInternalRow, Literal}
import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.internal.SQLConf
@@ -299,17 +299,20 @@ class CometCodegenFuzzSuite
case _ => false
}
spark.udf.register("id_int_arrdistinct", (i: Int) => i)
- for (field <- arrayStructFields) {
- val q = s"SELECT id_int_arrdistinct(cardinality(array_distinct(${field.name}))) FROM t1"
- val df = sql(q)
- val plan = df.queryExecution.optimizedPlan.toString
- val planLower = plan.toLowerCase
- assert(
- planLower.contains("array_distinct") || planLower.contains("arraydistinct"),
- s"optimizer eliminated array_distinct on column ${field.name}; coverage would be " +
- s"vacuous. plan=\n$plan")
- assertCodegenRan {
- checkSparkAnswerAndOperator(df)
+ // Inputs omit negative zeros and use Parquet-canonicalized NaNs for native opt-in coverage.
+ withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[ArrayDistinct]) -> "true") {
+ for (field <- arrayStructFields) {
+ val q = s"SELECT id_int_arrdistinct(cardinality(array_distinct(${field.name}))) FROM t1"
+ val df = sql(q)
+ val plan = df.queryExecution.optimizedPlan.toString
+ val planLower = plan.toLowerCase
+ assert(
+ planLower.contains("array_distinct") || planLower.contains("arraydistinct"),
+ s"optimizer eliminated array_distinct on column ${field.name}; coverage would be " +
+ s"vacuous. plan=\n$plan")
+ assertCodegenRan {
+ checkSparkAnswerAndOperator(df)
+ }
}
}
}
diff --git a/spark/src/test/scala/org/apache/comet/CometFloatSemanticsSuite.scala b/spark/src/test/scala/org/apache/comet/CometFloatSemanticsSuite.scala
index b7830cb35c..a5c679b609 100644
--- a/spark/src/test/scala/org/apache/comet/CometFloatSemanticsSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometFloatSemanticsSuite.scala
@@ -323,19 +323,6 @@ object CometFloatSemanticsSuite {
val cases: Seq[Case] =
comparisonCases ++ literalCases ++ keyCases ++ expressionCases ++ aggregateCases
- /**
- * SPARK-54918, in 4.0.5, 4.1.4 and 4.2.0, changed float equality in the array set functions.
- */
- private def hasSpark54918: Boolean = {
- val Array(major, minor, patch) =
- org.apache.spark.SPARK_VERSION.split("[.-]").take(3).map(_.toInt)
- (major, minor) match {
- case (4, 0) => patch >= 5
- case (4, 1) => patch >= 4
- case _ => major > 4 || (major == 4 && minor >= 2)
- }
- }
-
private def issue(n: Int): String = s"https://github.com/apache/datafusion-comet/issues/$n"
private def in(group: String, contexts: String*)(c: Case): Boolean =
@@ -374,11 +361,6 @@ object CometFloatSemanticsSuite {
c =>
in("expression", "greatest", "least")(c) ||
in("aggregate", "max", "min", "global max and min", "window max and min")(c)),
- KnownGap(
- issue(5701),
- "array_distinct and array_union fold -0.0 into 0.0 and keep NaNs with different bits " +
- "apart, where Spark before SPARK-54918 keeps the zeros apart and merges the NaNs.",
- c => in("expression", "array_distinct", "array_union")(c) && !hasSpark54918),
KnownGap(
issue(5312),
"collect_set before Spark 4.2 treats -0.0 and 0.0 as one value and NaNs as distinct.",
diff --git a/spark/src/test/scala/org/apache/comet/SqlFileTestParserSuite.scala b/spark/src/test/scala/org/apache/comet/SqlFileTestParserSuite.scala
index 64e0733900..c356c3c37a 100644
--- a/spark/src/test/scala/org/apache/comet/SqlFileTestParserSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/SqlFileTestParserSuite.scala
@@ -159,36 +159,75 @@ class SqlFileTestParserSuite extends AnyFunSuite {
assert(queries.map(_.sql) === Seq("SELECT abs(a) FROM t", "SELECT hypot(a, b) FROM t"))
}
- // #5702: NormalizeFloatingNumbers does not rewrite array-function inputs, so a
- // plain SELECT keeps -0.0 literals intact. The fixtures must skip those literal
- // cases for the same reason as the column-sourced ones, and must not claim that
- // Spark and Comet agree on the literal path.
- test("signed-zero array fixtures skip literals and do not claim Spark agreement") {
- val names =
- Seq("array_distinct.sql", "array_except.sql", "array_intersect.sql", "array_union.sql")
+ // #5702: Spark releases before SPARK-54918 keep -0.0 distinct in array_distinct and array_union,
+ // and native distinct/union run by default only on Spark 4.2.0, the one release that normalizes
+ // their arguments in the plan. The signed-zero cases therefore live in version-gated fixtures:
+ // the Spark 3 and 4.0/4.1 ones assert the fallback, and only the Spark 4.2+ one may claim that
+ // Comet runs them natively and agrees, through the native opt-ins.
+ test("signed-zero array fixtures are version-gated and claim agreement only on Spark 4.2+") {
+ def fixture(name: String): File = {
+ val url = getClass.getClassLoader.getResource(s"sql-tests/expressions/array/$name")
+ assert(url != null, s"missing fixture $name")
+ new File(url.toURI)
+ }
+ def queries(name: String): Seq[SqlQuery] =
+ SqlFileTestParser.parse(fixture(name)).records.collect { case q: SqlQuery => q }
+ def hasSignedZero(sql: String): Boolean = sql.contains("'-0.0'")
+ def isDistinctOrUnion(sql: String): Boolean =
+ sql.contains("array_distinct(") || sql.contains("array_union(")
+
val stalePhrases = Seq(
"both Spark and Comet collapse it and agree here",
"only rewrites literals, not parquet columns")
- names.foreach { name =>
- val url = getClass.getClassLoader.getResource(s"sql-tests/expressions/array/$name")
- assert(url != null, s"missing fixture $name")
- val file = new File(url.toURI)
- val text = {
- val src = scala.io.Source.fromFile(file, "UTF-8")
- try src.mkString
- finally src.close()
+ Seq("array_distinct.sql", "array_except.sql", "array_intersect.sql", "array_union.sql")
+ .foreach { name =>
+ val text = {
+ val src = scala.io.Source.fromFile(fixture(name), "UTF-8")
+ try src.mkString
+ finally src.close()
+ }
+ stalePhrases.foreach { phrase =>
+ assert(!text.contains(phrase), s"$name still claims: $phrase")
+ }
+ assert(
+ !text.contains("'-0.0'"),
+ s"$name has signed-zero cases outside the version-gated array_set_signed_zero fixtures")
}
- stalePhrases.foreach { phrase =>
- assert(!text.contains(phrase), s"$name still claims: $phrase")
+
+ // (fixture, MinSparkVersion, MaxSparkVersion, allowed mode for array_distinct/array_union)
+ val gated = Seq(
+ ("array_set_signed_zero_spark_3.sql", None, Some("3.5"), "expect_fallback"),
+ ("array_set_signed_zero_spark_4_0_4_1.sql", Some("4.0"), Some("4.1"), "expect_fallback"),
+ ("array_set_signed_zero.sql", Some("4.2"), None, "query"))
+ gated.foreach { case (name, minVersion, maxVersion, expectedMode) =>
+ val parsed = SqlFileTestParser.parse(fixture(name))
+ assert(parsed.minSparkVersion == minVersion, s"$name MinSparkVersion")
+ assert(parsed.maxSparkVersion == maxVersion, s"$name MaxSparkVersion")
+ if (expectedMode == "query") {
+ // 4.2.1+ falls back by default, so only the opt-ins keep this fixture patch-independent.
+ Seq("ArrayDistinct", "ArrayUnion").foreach { expr =>
+ assert(
+ parsed.configs.contains(CometConf.getExprAllowIncompatConfigKey(expr) -> "true"),
+ s"$name must opt $expr into native execution")
+ }
+ }
+ val cases = queries(name).filter(q => isDistinctOrUnion(q.sql))
+ Seq("array_distinct(", "array_union(").foreach { fn =>
+ assert(
+ cases.exists(q =>
+ q.sql.contains(fn) && hasSignedZero(q.sql) && !q.sql.toLowerCase.contains(" from ")),
+ s"$name is missing a signed-zero literal query for $fn")
}
- val ignoredLiterals = SqlFileTestParser.parse(file).records.collect {
- case SqlQuery(sql, Ignore(_), _)
- if sql.contains("array(") &&
- (sql.contains("double('-0.0')") || sql.contains("float('-0.0')")) &&
- !sql.toLowerCase.contains(" from ") =>
- sql
+ cases.foreach { q =>
+ val modeOk = (expectedMode, q.mode) match {
+ // Before 4.2, no release normalizes in the plan, so the native path must not run.
+ case ("expect_fallback", ExpectFallback(_)) => true
+ // Spark 4.2+ normalizes flat zeros like the native kernels, so they must run and agree.
+ case ("query", CheckCoverageAndAnswer) => true
+ case _ => false
+ }
+ assert(modeOk, s"$name line ${q.line}: ${q.mode} is not $expectedMode for ${q.sql}")
}
- assert(ignoredLiterals.nonEmpty, s"$name is missing an ignored signed-zero literal query")
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org