From c18b758169ed9f33c2af6dc4ed6edd93fdccb9e8 Mon Sep 17 00:00:00 2001 From: Viacheslav Inozemtsev Date: Sun, 30 Aug 2026 14:30:46 +0300 Subject: [PATCH 1/4] fix: return null from array_join when nullReplacement is null --- .../expression-audits/array_funcs.md | 3 +- docs/source/user-guide/latest/expressions.md | 2 +- .../scala/org/apache/comet/serde/arrays.scala | 33 ++++++++-- .../expressions/array/array_join.sql | 66 ++++++++++++++++++- .../array/array_join_null_replacement.sql | 52 +++++++++++++++ 5 files changed, 146 insertions(+), 10 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.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..e05a8358cbe 100644 --- a/docs/source/contributor-guide/expression-audits/array_funcs.md +++ b/docs/source/contributor-guide/expression-audits/array_funcs.md @@ -84,9 +84,10 @@ ## array_join - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8. -- Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayJoin(array, delimiter, nullReplacement)`. Comet routes via `CometArrayJoin` to DataFusion's `array_to_string` and is unconditionally flagged `Incompatible` ("Null handling may differ from Spark", [#3178](https://github.com/apache/datafusion-comet/issues/3178)). +- Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayJoin(array, delimiter, nullReplacement)`. Comet routes via `CometArrayJoin` to DataFusion's `array_to_string`. - Spark 4.0.1 (audited 2026-05-27): `inputTypes` widened to `AbstractArrayType(StringTypeWithCollation(supportsTrimCollation = true))`; non-binary collations not propagated ([#2190](https://github.com/apache/datafusion-comet/issues/2190)). - Spark 4.1.1 (audited 2026-05-27): adds `contextIndependentFoldable` override; runtime unchanged. +- Current status: `CometArrayJoin` reports `Compatible` for non-collated input. Spark's `ArrayJoin` returns null whenever `nullReplacement` evaluates to null, regardless of the array's contents, while DataFusion's `array_to_string` reads a null `null_string` as "omit null elements"; `convert` therefore wraps the three-argument call in an `IsNull(nullReplacement)` guard ([#3178](https://github.com/apache/datafusion-comet/issues/3178)). Null elements are skipped without a replacement and substituted with one, matching Spark in both cases. Non-default string collations remain `Incompatible` and route through the codegen dispatcher ([#2190](https://github.com/apache/datafusion-comet/issues/2190)). ## array_max diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index c8b75102340..2c7dd0a7ac2 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -155,7 +155,7 @@ The tables below list every Spark built-in expression with its current status. | `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)) | -| `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_join` | ✅ | Hybrid | Native by default; non-UTF8_BINARY collations route through the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | | `array_max` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | | `array_min` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | | `array_position` | ✅ | Native | Binary/struct/map/null elements fall back | 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..eeca7c97b55 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -22,7 +22,7 @@ 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, Cast, CreateArray, ElementAt, EmptyRow, Expression, Flatten, GetArrayItem, IsNotNull, LambdaFunction, Literal, NamedLambdaVariable, Reverse, Sequence, Size, Slice, SortArray, ZipWith} +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, Cast, CreateArray, ElementAt, EmptyRow, Expression, Flatten, GetArrayItem, IsNotNull, IsNull, LambdaFunction, Literal, NamedLambdaVariable, Reverse, Sequence, Size, Slice, SortArray, ZipWith} import org.apache.spark.sql.catalyst.util.GenericArrayData import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ @@ -335,13 +335,11 @@ object CometArrayJoin with CometTypeShim with CodegenDispatchFallback { - private val incompatReason = "Null handling may differ from Spark" - private val collationReason = "array_join does not propagate non-UTF8_BINARY collations to the output string " + "(https://github.com/apache/datafusion-comet/issues/2190)" - override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason, collationReason) + override def getIncompatibleReasons(): Seq[String] = Seq(collationReason) override def getSupportLevel(expr: ArrayJoin): SupportLevel = { // Spark 4.0 widens ArrayJoin's input to StringTypeWithCollation. Concatenation itself is @@ -352,7 +350,9 @@ object CometArrayJoin if (hasNonDefaultStringCollation(expr.array.dataType)) { Incompatible(Some(collationReason)) } else { - Incompatible(Some(incompatReason)) + // Null handling matched Spark once the nullReplacement guard in convert() landed (#3178); + // collation is the only remaining deviation. + Compatible() } } @@ -374,7 +374,28 @@ object CometArrayJoin delimiterExprProto, nullReplacementExprProto) - arrayJoinScalarExpr + // Spark's ArrayJoin returns null as soon as nullReplacement evaluates to null, whether + // or not the array actually contains any nulls. DataFusion's array_to_string instead + // reads a null null_string as "omit null elements", which is what the two-argument form + // means, so wrap the call in an explicit null guard (#3178). A non-nullable replacement + // -- the common literal case -- cannot trigger this and is left unwrapped. + if (!nullReplacementExpr.nullable) { + arrayJoinScalarExpr + } else { + for { + joined <- arrayJoinScalarExpr + replacementIsNull <- exprToProto(IsNull(nullReplacementExpr), inputs, binding) + nullLiteral <- exprToProto(Literal(null, expr.dataType), inputs, binding) + } yield ExprOuterClass.Expr + .newBuilder() + .setIf( + ExprOuterClass.IfExpr + .newBuilder() + .setIfExpr(replacementIsNull) + .setTrueExpr(nullLiteral) + .setFalseExpr(joined)) + .build() + } case None => val arrayJoinScalarExpr = scalarFunctionExprToProto("array_to_string", arrayExprProto, delimiterExprProto) diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join.sql index ca81a918d89..3bcc0d238b2 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_join.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_join.sql @@ -15,14 +15,76 @@ -- specific language governing permissions and limitations -- under the License. +-- array_join runs natively by default (CometArrayJoin reports Compatible for non-collated +-- input); collated input still routes through the JVM codegen dispatcher and is covered by +-- array_join_collation.sql. +-- +-- Spark's array_join semantics (#3178): +-- * without nullReplacement, null elements are skipped entirely +-- * with nullReplacement, null elements are replaced by that string +-- * a null array or a null delimiter yields null +-- * an empty array yields the empty string +-- Null placement matters: a leading or trailing null must not leave a dangling delimiter. + statement -CREATE TABLE test_array_join(arr array) USING parquet +CREATE TABLE test_array_join(arr array, delim string, nullrep string) USING parquet statement -INSERT INTO test_array_join VALUES (array('a', 'b', 'c')), (array('hello', 'world')), (array()), (NULL), (array('a', NULL, 'c')) +INSERT INTO test_array_join VALUES + (array('a', 'b', 'c'), ',', 'N'), + (array('hello', 'world'), ' ', 'N'), + (array(), ',', 'N'), + (NULL, ',', 'N'), + (array('a', NULL, 'c'), ',', 'N'), + (array(NULL, 'b', 'c'), ',', 'N'), + (array('a', 'b', NULL), ',', 'N'), + (array(NULL), ',', 'N'), + (array(NULL, NULL), ',', 'N'), + (array('', 'b'), ',', 'N'), + (array('a', 'b'), NULL, 'N') +-- column array, literal delimiter, no null replacement (nulls skipped) query SELECT array_join(arr, ',') FROM test_array_join +-- column array, literal delimiter and null replacement (nulls replaced) query SELECT array_join(arr, ',', 'NULL') FROM test_array_join + +-- all three arguments as columns, including null delimiter and null replacement rows +query +SELECT array_join(arr, delim, nullrep) FROM test_array_join + +-- column array with a column delimiter but a literal replacement +query +SELECT array_join(arr, delim, 'N') FROM test_array_join + +-- multi-character and empty delimiters +query +SELECT array_join(arr, ' -- '), array_join(arr, '') FROM test_array_join + +-- the exact cases named in #3178, as literals (constant folding is disabled by the suite) +query +SELECT array_join(array('a', NULL, 'b'), ','), array_join(array('a', NULL, 'b'), ',', 'X') + +query +SELECT array_join(array('hello', NULL, 'world'), ' '), array_join(array('hello', NULL, 'world'), ' ', 'NULL') + +-- literal null placement: leading, trailing, only-null, all-null +query +SELECT array_join(array(NULL, 'b'), ','), array_join(array('a', NULL), ','), array_join(cast(array(NULL) as array), ','), array_join(cast(array(NULL, NULL) as array), ',') + +query +SELECT array_join(array(NULL, 'b'), ',', 'X'), array_join(array('a', NULL), ',', 'X'), array_join(cast(array(NULL) as array), ',', 'X'), array_join(cast(array(NULL, NULL) as array), ',', 'X') + +-- empty string elements are not nulls and must survive both paths +query +SELECT array_join(array('', 'b'), ','), array_join(array('', NULL, 'b'), ','), array_join(array('', NULL, 'b'), ',', 'X') + +-- empty array and null array as literals +query +SELECT array_join(cast(array() as array), ','), array_join(cast(NULL as array), ','), array_join(cast(array() as array), ',', 'X'), array_join(cast(NULL as array), ',', 'X') + +-- null delimiter as a literal +query +SELECT array_join(array('a', 'b'), cast(NULL as string)) diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql new file mode 100644 index 00000000000..52181a5c5d8 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql @@ -0,0 +1,52 @@ +-- 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 coverage for #3178. +-- +-- Spark's ArrayJoin short-circuits to null on three conditions: a null array, a null delimiter, +-- and a null nullReplacement. DataFusion's array_to_string only short-circuits on the first two; +-- a null null_string collapses to None, which it reads as "omit null elements" -- the same as +-- not passing a third argument at all. CometArrayJoin therefore wraps the three-argument call in +-- an explicit null guard. Note the third case below: a null replacement must nullify the result +-- even when the array contains no nulls for it to replace. + +statement +CREATE TABLE test_aj_nullrep(arr array, delim string, nullrep string) USING parquet + +statement +INSERT INTO test_aj_nullrep VALUES + (array('a', NULL, 'c'), ',', NULL), + (array('a', 'b', 'c'), ',', NULL), + (array(NULL, NULL), ',', NULL), + (NULL, ',', NULL), + (array('a', NULL, 'c'), ',', 'X') + +-- null replacement as a column, mixed with a non-null replacement row +query +SELECT array_join(arr, delim, nullrep) FROM test_aj_nullrep + +-- null replacement as a literal, array containing nulls +query +SELECT array_join(array('a', NULL, 'b'), ',', cast(NULL as string)) + +-- null replacement as a literal, array containing no nulls at all +query +SELECT array_join(array('a', 'b'), ',', cast(NULL as string)) + +-- a non-nullable literal replacement takes the unwrapped path +query +SELECT array_join(array('a', NULL, 'b'), ',', 'X') From b34f5803a932d688f878820001862b4338bc8826 Mon Sep 17 00:00:00 2001 From: Viacheslav Inozemtsev Date: Wed, 2 Sep 2026 19:06:10 +0300 Subject: [PATCH 2/4] fix: guard array_join arguments to preserve Spark's null short-circuiting --- .../expression-audits/array_funcs.md | 2 +- docs/source/user-guide/latest/expressions.md | 2 +- .../scala/org/apache/comet/serde/arrays.scala | 91 ++++++++++------- .../expressions/array/array_join.sql | 25 +++-- .../array/array_join_null_array_guard.sql | 41 ++++++++ .../array/array_join_null_replacement.sql | 11 +-- .../comet/CometArrayExpressionSuite.scala | 97 +++++++++++++++---- 7 files changed, 196 insertions(+), 73 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_join_null_array_guard.sql diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index e05a8358cbe..180477403f5 100644 --- a/docs/source/contributor-guide/expression-audits/array_funcs.md +++ b/docs/source/contributor-guide/expression-audits/array_funcs.md @@ -87,7 +87,7 @@ - Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayJoin(array, delimiter, nullReplacement)`. Comet routes via `CometArrayJoin` to DataFusion's `array_to_string`. - Spark 4.0.1 (audited 2026-05-27): `inputTypes` widened to `AbstractArrayType(StringTypeWithCollation(supportsTrimCollation = true))`; non-binary collations not propagated ([#2190](https://github.com/apache/datafusion-comet/issues/2190)). - Spark 4.1.1 (audited 2026-05-27): adds `contextIndependentFoldable` override; runtime unchanged. -- Current status: `CometArrayJoin` reports `Compatible` for non-collated input. Spark's `ArrayJoin` returns null whenever `nullReplacement` evaluates to null, regardless of the array's contents, while DataFusion's `array_to_string` reads a null `null_string` as "omit null elements"; `convert` therefore wraps the three-argument call in an `IsNull(nullReplacement)` guard ([#3178](https://github.com/apache/datafusion-comet/issues/3178)). Null elements are skipped without a replacement and substituted with one, matching Spark in both cases. Non-default string collations remain `Incompatible` and route through the codegen dispatcher ([#2190](https://github.com/apache/datafusion-comet/issues/2190)). +- Current status: `CometArrayJoin` reports `Compatible` for non-collated input. `convert` nests the `array_to_string` call inside `IsNull` guards so that a null array, delimiter or `nullReplacement` short-circuits the way Spark's does ([#3178](https://github.com/apache/datafusion-comet/issues/3178)). A guarded argument is serialized twice, so non-deterministic ones are reported `Incompatible` and run through the codegen dispatcher, as do non-default string collations ([#2190](https://github.com/apache/datafusion-comet/issues/2190)). ## array_max diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 2c7dd0a7ac2..97fefe7405f 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -155,7 +155,7 @@ The tables below list every Spark built-in expression with its current status. | `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)) | -| `array_join` | ✅ | Hybrid | Native by default; non-UTF8_BINARY collations route through the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | +| `array_join` | ✅ | Hybrid | Native by default; non-UTF8_BINARY collations and non-deterministic arguments route through the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | | `array_max` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | | `array_min` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | | `array_position` | ✅ | Native | Binary/struct/map/null elements fall back | 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 eeca7c97b55..65b3a15c2ef 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -339,7 +339,34 @@ object CometArrayJoin "array_join does not propagate non-UTF8_BINARY collations to the output string " + "(https://github.com/apache/datafusion-comet/issues/2190)" - override def getIncompatibleReasons(): Seq[String] = Seq(collationReason) + private val nonDeterministicReason = + "array_join would have to evaluate a non-deterministic argument twice to reproduce Spark's " + + "null short-circuiting (https://github.com/apache/datafusion-comet/issues/3178)" + + /** + * Arguments needing an `IsNull` guard around the native call, in Spark's evaluation order. + * + * Spark short-circuits to null on the array, then the delimiter, then the null replacement, and + * never evaluates the later arguments; DataFusion evaluates all of them eagerly. `IfExpr` is a + * DataFusion `CaseExpr`, which evaluates branches against a filtered batch, so nesting restores + * that ordering. Only a non-foldable later argument needs protecting. The null replacement is + * guarded whenever it is nullable: `array_to_string` reads a null `null_string` as "omit nulls" + * rather than nullifying the row (#3178). + */ + private def guardedArgs(expr: ArrayJoin): Seq[Expression] = { + val afterArray = expr.delimiter +: expr.nullReplacement.toSeq + val arrayGuard = + if (expr.array.nullable && afterArray.exists(!_.foldable)) Seq(expr.array) else Nil + val delimiterGuard = + if (expr.delimiter.nullable && expr.nullReplacement.exists(!_.foldable)) { + Seq(expr.delimiter) + } else Nil + val replacementGuard = expr.nullReplacement.filter(_.nullable).toSeq + arrayGuard ++ delimiterGuard ++ replacementGuard + } + + override def getIncompatibleReasons(): Seq[String] = + Seq(collationReason, nonDeterministicReason) override def getSupportLevel(expr: ArrayJoin): SupportLevel = { // Spark 4.0 widens ArrayJoin's input to StringTypeWithCollation. Concatenation itself is @@ -349,9 +376,11 @@ object CometArrayJoin // array_join native and matching Spark, consistent with CometReverse's #2190 handling. if (hasNonDefaultStringCollation(expr.array.dataType)) { Incompatible(Some(collationReason)) + } else if (guardedArgs(expr).exists(!_.deterministic)) { + // A guarded argument is serialized twice and CaseExpr evaluates the false branch on a + // filtered batch, so a non-deterministic one would diverge between the two copies. + Incompatible(Some(nonDeterministicReason)) } else { - // Null handling matched Spark once the nullReplacement guard in convert() landed (#3178); - // collation is the only remaining deviation. Compatible() } } @@ -360,47 +389,35 @@ object CometArrayJoin expr: ArrayJoin, inputs: Seq[Attribute], binding: Boolean): Option[ExprOuterClass.Expr] = { - val arrayExpr = expr.asInstanceOf[ArrayJoin] - val arrayExprProto = exprToProto(arrayExpr.array, inputs, binding) - val delimiterExprProto = exprToProto(arrayExpr.delimiter, inputs, binding) + val arrayExprProto = exprToProto(expr.array, inputs, binding) + val delimiterExprProto = exprToProto(expr.delimiter, inputs, binding) - arrayExpr.nullReplacement match { + val joined = expr.nullReplacement match { case Some(nullReplacementExpr) => - val nullReplacementExprProto = exprToProto(nullReplacementExpr, inputs, binding) - - val arrayJoinScalarExpr = scalarFunctionExprToProto( + scalarFunctionExprToProto( "array_to_string", arrayExprProto, delimiterExprProto, - nullReplacementExprProto) - - // Spark's ArrayJoin returns null as soon as nullReplacement evaluates to null, whether - // or not the array actually contains any nulls. DataFusion's array_to_string instead - // reads a null null_string as "omit null elements", which is what the two-argument form - // means, so wrap the call in an explicit null guard (#3178). A non-nullable replacement - // -- the common literal case -- cannot trigger this and is left unwrapped. - if (!nullReplacementExpr.nullable) { - arrayJoinScalarExpr - } else { - for { - joined <- arrayJoinScalarExpr - replacementIsNull <- exprToProto(IsNull(nullReplacementExpr), inputs, binding) - nullLiteral <- exprToProto(Literal(null, expr.dataType), inputs, binding) - } yield ExprOuterClass.Expr - .newBuilder() - .setIf( - ExprOuterClass.IfExpr - .newBuilder() - .setIfExpr(replacementIsNull) - .setTrueExpr(nullLiteral) - .setFalseExpr(joined)) - .build() - } + exprToProto(nullReplacementExpr, inputs, binding)) case None => - val arrayJoinScalarExpr = - scalarFunctionExprToProto("array_to_string", arrayExprProto, delimiterExprProto) + scalarFunctionExprToProto("array_to_string", arrayExprProto, delimiterExprProto) + } - arrayJoinScalarExpr + // The outermost guard is the one Spark evaluates first. + guardedArgs(expr).foldRight(joined) { (arg, inner) => + for { + innerProto <- inner + argIsNull <- exprToProto(IsNull(arg), inputs, binding) + nullLiteral <- exprToProto(Literal(null, expr.dataType), inputs, binding) + } yield ExprOuterClass.Expr + .newBuilder() + .setIf( + ExprOuterClass.IfExpr + .newBuilder() + .setIfExpr(argIsNull) + .setTrueExpr(nullLiteral) + .setFalseExpr(innerProto)) + .build() } } } diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join.sql index 3bcc0d238b2..05c87257295 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_join.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_join.sql @@ -15,16 +15,10 @@ -- specific language governing permissions and limitations -- under the License. --- array_join runs natively by default (CometArrayJoin reports Compatible for non-collated --- input); collated input still routes through the JVM codegen dispatcher and is covered by +-- Spark skips null elements without a nullReplacement and substitutes them with one; a null +-- array or delimiter yields null and an empty array the empty string. Null placement matters: +-- a leading or trailing null must not leave a dangling delimiter. Collated input is covered by -- array_join_collation.sql. --- --- Spark's array_join semantics (#3178): --- * without nullReplacement, null elements are skipped entirely --- * with nullReplacement, null elements are replaced by that string --- * a null array or a null delimiter yields null --- * an empty array yields the empty string --- Null placement matters: a leading or trailing null must not leave a dangling delimiter. statement CREATE TABLE test_array_join(arr array, delim string, nullrep string) USING parquet @@ -88,3 +82,16 @@ SELECT array_join(cast(array() as array), ','), array_join(cast(NULL as -- null delimiter as a literal query SELECT array_join(array('a', 'b'), cast(NULL as string)) + +-- Spark's inputTypes accepts any array that implicitly casts to array, so non-string +-- element types are valid and common in practice. +query +SELECT array_join(array(1, 2, 3), ','), array_join(array(1, NULL, 3), ','), array_join(array(1, NULL, 3), ',', 'X') + +query +SELECT array_join(array(1.5, NULL, 2.5), ',', 'X'), array_join(array(true, NULL, false), ',', 'X') + +-- an empty-string replacement is not a null replacement; '' substitutes, NULL nullifies, and that +-- is exactly the distinction the null guard draws +query +SELECT array_join(array('a', NULL, 'b'), ',', ''), array_join(array('a', NULL, 'b'), '', '') diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join_null_array_guard.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join_null_array_guard.sql new file mode 100644 index 00000000000..e95e8157ddc --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_join_null_array_guard.sql @@ -0,0 +1,41 @@ +-- 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. + +-- Spark short-circuits on a null array before evaluating the delimiter, while DataFusion +-- evaluates every argument eagerly. element_at is 1-based, so delims[0] throws whenever it is +-- evaluated: the arr IS NULL row must not reach it (#3178). + +statement +CREATE TABLE test_aj_guard(arr array, delims array, idx int) USING parquet + +statement +INSERT INTO test_aj_guard VALUES + (NULL, array(','), 0), + (array('a', 'b'), array(','), 1), + (array('a', NULL, 'b'), array(';'), 1) + +-- the null-array row short-circuits before element_at(delims, 0) can throw +query +SELECT array_join(arr, element_at(delims, idx)) FROM test_aj_guard WHERE arr IS NULL + +-- rows with a real index still join normally +query +SELECT array_join(arr, element_at(delims, idx)) FROM test_aj_guard WHERE arr IS NOT NULL + +-- and the same shape with a null replacement +query +SELECT array_join(arr, element_at(delims, idx), 'X') FROM test_aj_guard WHERE arr IS NULL diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql index 52181a5c5d8..2d9a77d8387 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql @@ -15,14 +15,9 @@ -- specific language governing permissions and limitations -- under the License. --- Regression coverage for #3178. --- --- Spark's ArrayJoin short-circuits to null on three conditions: a null array, a null delimiter, --- and a null nullReplacement. DataFusion's array_to_string only short-circuits on the first two; --- a null null_string collapses to None, which it reads as "omit null elements" -- the same as --- not passing a third argument at all. CometArrayJoin therefore wraps the three-argument call in --- an explicit null guard. Note the third case below: a null replacement must nullify the result --- even when the array contains no nulls for it to replace. +-- Regression coverage for #3178. Spark returns null whenever nullReplacement is null, even for +-- an array with no nulls to replace, while array_to_string reads a null null_string as "omit +-- nulls". statement CREATE TABLE test_aj_nullrep(arr array, delim string, nullrep string) USING parquet diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index 05a6e8e650d..d89e9e9d712 100644 --- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala @@ -25,14 +25,15 @@ 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.{ArrayContains, ArrayRemove} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, If, Literal, MonotonicallyIncreasingID, Remainder} 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 +import org.apache.spark.sql.types.{ArrayType, StringType} import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus} import org.apache.comet.DataTypeSupport.isComplexType -import org.apache.comet.serde.{CometArrayExcept, CometArrayRemove, CometArrayReverse, CometFlatten} +import org.apache.comet.serde.{CometArrayExcept, CometArrayJoin, CometArrayRemove, CometArrayReverse, CometFlatten, Compatible, Incompatible} import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions} class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { @@ -518,28 +519,90 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp } } + // No allowIncompatible opt-in: array_join runs natively by default now. test("array_join") { - withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[ArrayJoin]) -> "true") { - Seq(true, false).foreach { dictionaryEnabled => - withTempDir { dir => - withTempView("t1") { - val path = new Path(dir.toURI.toString, "test.parquet") - makeParquetFileAllPrimitiveTypes(path, dictionaryEnabled, 10000) - spark.read.parquet(path.toString).createOrReplaceTempView("t1") - checkSparkAnswerAndOperator(sql( - "SELECT array_join(array(cast(_1 as string), cast(_2 as string), cast(_6 as string)), ' @ ') from t1")) - checkSparkAnswerAndOperator(sql( - "SELECT array_join(array(cast(_1 as string), cast(_2 as string), cast(_6 as string)), ' @ ', ' +++ ') from t1")) - checkSparkAnswerAndOperator(sql( - "SELECT array_join(array('hello', 'world', cast(_2 as string)), ' ') from t1 where _2 is not null")) - checkSparkAnswerAndOperator(sql( + Seq(true, false).foreach { dictionaryEnabled => + withTempDir { dir => + withTempView("t1") { + val path = new Path(dir.toURI.toString, "test.parquet") + makeParquetFileAllPrimitiveTypes(path, dictionaryEnabled, 10000) + spark.read.parquet(path.toString).createOrReplaceTempView("t1") + checkSparkAnswerAndOperator(sql( + "SELECT array_join(array(cast(_1 as string), cast(_2 as string), cast(_6 as string)), ' @ ') from t1")) + checkSparkAnswerAndOperator(sql( + "SELECT array_join(array(cast(_1 as string), cast(_2 as string), cast(_6 as string)), ' @ ', ' +++ ') from t1")) + checkSparkAnswerAndOperator(sql( + "SELECT array_join(array('hello', 'world', cast(_2 as string)), ' ') from t1 where _2 is not null")) + checkSparkAnswerAndOperator( + sql( "SELECT array_join(array('hello', '-', 'world', cast(_2 as string)), ' ') from t1")) - } + // nullable, non-foldable delimiter and null replacement: the guarded shape + checkSparkAnswerAndOperator( + sql("SELECT array_join(array('a', cast(_2 as string), 'b'), cast(_6 as string), " + + "cast(_1 as string)) from t1")) } } } } + // Result assertions cannot tell the native path from the codegen dispatcher, since an + // Incompatible verdict runs Spark's own doGenCode and matches. Pin the verdict itself. + test("array_join support level pins the native path") { + val nullableArray = AttributeReference("arr", ArrayType(StringType), nullable = true)() + val nullableStr = AttributeReference("s", StringType, nullable = true)() + + // literal delimiter, no replacement + assert( + CometArrayJoin + .getSupportLevel(ArrayJoin(nullableArray, Literal(","), None)) + .isInstanceOf[Compatible]) + // literal delimiter and literal replacement + assert( + CometArrayJoin + .getSupportLevel(ArrayJoin(nullableArray, Literal(","), Some(Literal("X")))) + .isInstanceOf[Compatible]) + // nullable column replacement still takes the native path, behind the guard + assert( + CometArrayJoin + .getSupportLevel(ArrayJoin(nullableArray, Literal(","), Some(nullableStr))) + .isInstanceOf[Compatible]) + + // A guarded non-deterministic argument would be evaluated twice. + val nonDeterministicReplacement = + If( + EqualTo(Remainder(MonotonicallyIncreasingID(), Literal(2L)), Literal(0L)), + Literal.create(null, StringType), + Literal("X")) + assert( + CometArrayJoin + .getSupportLevel( + ArrayJoin(nullableArray, Literal(","), Some(nonDeterministicReplacement))) + .isInstanceOf[Incompatible]) + } + + test("array_join emits a null guard only where it is needed") { + val nullableArray = AttributeReference("arr", ArrayType(StringType), nullable = true)() + val nullableStr = AttributeReference("s", StringType, nullable = true)() + val inputs = Seq(nullableArray, nullableStr) + + def convert(expr: ArrayJoin) = CometArrayJoin.convert(expr, inputs, binding = false) + + // All-foldable trailing arguments: no guard, so the plan is unchanged. + val unguarded = convert(ArrayJoin(nullableArray, Literal(","), Some(Literal("X")))) + assert(unguarded.isDefined) + assert(!unguarded.get.hasIf, "no guard should be emitted for a non-nullable replacement") + + // Nullable replacement: Spark nullifies the row. + val guarded = convert(ArrayJoin(nullableArray, Literal(","), Some(nullableStr))) + assert(guarded.isDefined) + assert(guarded.get.hasIf, "a nullable replacement must be guarded") + + // Non-foldable delimiter: Spark never evaluates it when the array is null. + val arrayGuarded = convert(ArrayJoin(nullableArray, nullableStr, None)) + assert(arrayGuarded.isDefined) + assert(arrayGuarded.get.hasIf, "a non-foldable delimiter must put the array behind a guard") + } + test("arrays_overlap") { Seq(true, false).foreach { dictionaryEnabled => withTempDir { dir => From cb86c8484c01c0e786bf1d5fcb223845bbceeaf8 Mon Sep 17 00:00:00 2001 From: Viacheslav Inozemtsev Date: Thu, 3 Sep 2026 11:14:02 +0300 Subject: [PATCH 3/4] fix: restrict array_join's native path to order-insensitive arguments --- .../expression-audits/array_funcs.md | 2 +- docs/source/user-guide/latest/expressions.md | 2 +- .../scala/org/apache/comet/serde/arrays.scala | 76 +++++++--------- .../expressions/array/array_join.sql | 5 +- .../array/array_join_eager_eval_dispatch.sql | 48 ++++++++++ .../array/array_join_null_array_guard.sql | 41 --------- .../array/array_join_null_replacement.sql | 17 ++-- .../comet/CometArrayExpressionSuite.scala | 91 +++++++++---------- 8 files changed, 137 insertions(+), 145 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_join_eager_eval_dispatch.sql delete mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_join_null_array_guard.sql diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index 180477403f5..b4ae0b5b3a1 100644 --- a/docs/source/contributor-guide/expression-audits/array_funcs.md +++ b/docs/source/contributor-guide/expression-audits/array_funcs.md @@ -87,7 +87,7 @@ - Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayJoin(array, delimiter, nullReplacement)`. Comet routes via `CometArrayJoin` to DataFusion's `array_to_string`. - Spark 4.0.1 (audited 2026-05-27): `inputTypes` widened to `AbstractArrayType(StringTypeWithCollation(supportsTrimCollation = true))`; non-binary collations not propagated ([#2190](https://github.com/apache/datafusion-comet/issues/2190)). - Spark 4.1.1 (audited 2026-05-27): adds `contextIndependentFoldable` override; runtime unchanged. -- Current status: `CometArrayJoin` reports `Compatible` for non-collated input. `convert` nests the `array_to_string` call inside `IsNull` guards so that a null array, delimiter or `nullReplacement` short-circuits the way Spark's does ([#3178](https://github.com/apache/datafusion-comet/issues/3178)). A guarded argument is serialized twice, so non-deterministic ones are reported `Incompatible` and run through the codegen dispatcher, as do non-default string collations ([#2190](https://github.com/apache/datafusion-comet/issues/2190)). +- Current status: `CometArrayJoin` reports `Compatible` when the delimiter and null replacement are literals or column reads; Spark short-circuits past those arguments and DataFusion does not, so anything else runs through the codegen dispatcher, as do non-default string collations ([#2190](https://github.com/apache/datafusion-comet/issues/2190)). A nullable replacement is wrapped in an `IsNull` guard, since `array_to_string` reads a null `null_string` as "omit nulls" ([#3178](https://github.com/apache/datafusion-comet/issues/3178)). ## array_max diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index bc12fad5565..12c7f47f069 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -155,7 +155,7 @@ The tables below list every Spark built-in expression with its current status. | `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)) | -| `array_join` | ✅ | Hybrid | Native by default; non-UTF8_BINARY collations and non-deterministic arguments route through the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | +| `array_join` | ✅ | Hybrid | Native for literal or column delimiter and null replacement; other cases and non-UTF8_BINARY collations use the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | | `array_max` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | | `array_min` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | | `array_position` | ✅ | Native | Binary/struct/map/null elements fall back | 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 55daa2126e2..0181aa62418 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -22,7 +22,7 @@ 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, Cast, CreateArray, ElementAt, EmptyRow, Expression, Flatten, GetArrayItem, IsNotNull, IsNull, LambdaFunction, Literal, NamedLambdaVariable, Reverse, Sequence, Size, Slice, SortArray, ZipWith} +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, Size, Slice, SortArray, ZipWith} import org.apache.spark.sql.catalyst.util.GenericArrayData import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ @@ -361,34 +361,25 @@ object CometArrayJoin "array_join does not propagate non-UTF8_BINARY collations to the output string " + "(https://github.com/apache/datafusion-comet/issues/2190)" - private val nonDeterministicReason = - "array_join would have to evaluate a non-deterministic argument twice to reproduce Spark's " + - "null short-circuiting (https://github.com/apache/datafusion-comet/issues/3178)" + private val eagerEvalReason = + "array_join evaluates its delimiter and null replacement eagerly, while Spark short-circuits " + + "past them (https://github.com/apache/datafusion-comet/issues/3178)" /** - * Arguments needing an `IsNull` guard around the native call, in Spark's evaluation order. + * Whether evaluating `expr` earlier than Spark would is unobservable. * - * Spark short-circuits to null on the array, then the delimiter, then the null replacement, and - * never evaluates the later arguments; DataFusion evaluates all of them eagerly. `IfExpr` is a - * DataFusion `CaseExpr`, which evaluates branches against a filtered batch, so nesting restores - * that ordering. Only a non-foldable later argument needs protecting. The null replacement is - * guarded whenever it is nullable: `array_to_string` reads a null `null_string` as "omit nulls" - * rather than nullifying the row (#3178). + * Spark skips ArrayJoin's later arguments once an earlier one is null, and `eval` and + * `doGenCode` disagree on that order, while DataFusion evaluates every argument up front. A + * literal or column read cannot throw, carry state or have a side effect, so ordering cannot be + * observed for it; anything else goes to the codegen dispatcher. `foldable` is not usable here: + * ConstantFolding leaves a throwing foldable expression unfolded in a conditional branch. */ - private def guardedArgs(expr: ArrayJoin): Seq[Expression] = { - val afterArray = expr.delimiter +: expr.nullReplacement.toSeq - val arrayGuard = - if (expr.array.nullable && afterArray.exists(!_.foldable)) Seq(expr.array) else Nil - val delimiterGuard = - if (expr.delimiter.nullable && expr.nullReplacement.exists(!_.foldable)) { - Seq(expr.delimiter) - } else Nil - val replacementGuard = expr.nullReplacement.filter(_.nullable).toSeq - arrayGuard ++ delimiterGuard ++ replacementGuard + private def orderInsensitive(expr: Expression): Boolean = expr match { + case _: Literal | _: Attribute | _: BoundReference => true + case _ => false } - override def getIncompatibleReasons(): Seq[String] = - Seq(collationReason, nonDeterministicReason) + override def getIncompatibleReasons(): Seq[String] = Seq(collationReason, eagerEvalReason) override def getSupportLevel(expr: ArrayJoin): SupportLevel = { // Spark 4.0 widens ArrayJoin's input to StringTypeWithCollation. Concatenation itself is @@ -398,10 +389,8 @@ object CometArrayJoin // array_join native and matching Spark, consistent with CometReverse's #2190 handling. if (hasNonDefaultStringCollation(expr.array.dataType)) { Incompatible(Some(collationReason)) - } else if (guardedArgs(expr).exists(!_.deterministic)) { - // A guarded argument is serialized twice and CaseExpr evaluates the false branch on a - // filtered batch, so a non-deterministic one would diverge between the two copies. - Incompatible(Some(nonDeterministicReason)) + } else if (!(expr.delimiter +: expr.nullReplacement.toSeq).forall(orderInsensitive)) { + Incompatible(Some(eagerEvalReason)) } else { Compatible() } @@ -425,21 +414,24 @@ object CometArrayJoin scalarFunctionExprToProto("array_to_string", arrayExprProto, delimiterExprProto) } - // The outermost guard is the one Spark evaluates first. - guardedArgs(expr).foldRight(joined) { (arg, inner) => - for { - innerProto <- inner - argIsNull <- exprToProto(IsNull(arg), inputs, binding) - nullLiteral <- exprToProto(Literal(null, expr.dataType), inputs, binding) - } yield ExprOuterClass.Expr - .newBuilder() - .setIf( - ExprOuterClass.IfExpr - .newBuilder() - .setIfExpr(argIsNull) - .setTrueExpr(nullLiteral) - .setFalseExpr(innerProto)) - .build() + // Spark returns null as soon as nullReplacement is null, whatever the array holds, while + // array_to_string reads a null null_string as "omit nulls" (#3178). + expr.nullReplacement.filter(_.nullable) match { + case Some(nullReplacementExpr) => + for { + innerProto <- joined + replacementIsNull <- exprToProto(IsNull(nullReplacementExpr), inputs, binding) + nullLiteral <- exprToProto(Literal(null, expr.dataType), inputs, binding) + } yield ExprOuterClass.Expr + .newBuilder() + .setIf( + ExprOuterClass.IfExpr + .newBuilder() + .setIfExpr(replacementIsNull) + .setTrueExpr(nullLiteral) + .setFalseExpr(innerProto)) + .build() + case None => joined } } } diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join.sql index 05c87257295..533b529fae2 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_join.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_join.sql @@ -17,8 +17,7 @@ -- Spark skips null elements without a nullReplacement and substitutes them with one; a null -- array or delimiter yields null and an empty array the empty string. Null placement matters: --- a leading or trailing null must not leave a dangling delimiter. Collated input is covered by --- array_join_collation.sql. +-- a leading or trailing null must not leave a dangling delimiter. statement CREATE TABLE test_array_join(arr array, delim string, nullrep string) USING parquet @@ -71,7 +70,7 @@ SELECT array_join(array(NULL, 'b'), ','), array_join(array('a', NULL), ','), arr query SELECT array_join(array(NULL, 'b'), ',', 'X'), array_join(array('a', NULL), ',', 'X'), array_join(cast(array(NULL) as array), ',', 'X'), array_join(cast(array(NULL, NULL) as array), ',', 'X') --- empty string elements are not nulls and must survive both paths +-- empty string elements are not nulls query SELECT array_join(array('', 'b'), ','), array_join(array('', NULL, 'b'), ','), array_join(array('', NULL, 'b'), ',', 'X') diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join_eager_eval_dispatch.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join_eager_eval_dispatch.sql new file mode 100644 index 00000000000..b934fa43f1c --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_join_eager_eval_dispatch.sql @@ -0,0 +1,48 @@ +-- 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. + +-- A delimiter or replacement that can throw or carry state is routed to the codegen dispatcher, +-- because DataFusion evaluates every argument up front while Spark short-circuits past them +-- (#3178). These must return Spark's answers rather than raising INVALID_INDEX_OF_ZERO. + +statement +CREATE TABLE test_aj_eager(arr array, delims array, nullrep string, flag boolean) USING parquet + +statement +INSERT INTO test_aj_eager VALUES + (NULL, array(','), NULL, true), + (array('a', 'b'), array(','), NULL, true), + (array('a', NULL, 'b'), array(','), 'X', true), + (array('a', 'b'), array(','), 'X', false) + +-- element_at is 1-based, so index 0 throws whenever it is evaluated. A *foldable* throwing +-- delimiter is pinned in CometArrayExpressionSuite instead: the dispatcher cannot compile that +-- shape today, and the same failure reproduces on main. +query +SELECT array_join(arr, element_at(delims, 0)) FROM test_aj_eager WHERE arr IS NULL + +-- doGenCode evaluates the replacement before the delimiter, so a null replacement wins +query +SELECT array_join(arr, element_at(delims, 0), nullrep) FROM test_aj_eager WHERE nullrep IS NULL + +-- a non-deterministic replacement is evaluated once per row by Spark +query +SELECT array_join(arr, ',', cast(monotonically_increasing_id() as string)) IS NOT NULL FROM test_aj_eager WHERE arr IS NOT NULL + +-- the rows that do join still produce the right answer +query +SELECT array_join(arr, element_at(delims, 1), nullrep) FROM test_aj_eager WHERE arr IS NOT NULL diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join_null_array_guard.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join_null_array_guard.sql deleted file mode 100644 index e95e8157ddc..00000000000 --- a/spark/src/test/resources/sql-tests/expressions/array/array_join_null_array_guard.sql +++ /dev/null @@ -1,41 +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. - --- Spark short-circuits on a null array before evaluating the delimiter, while DataFusion --- evaluates every argument eagerly. element_at is 1-based, so delims[0] throws whenever it is --- evaluated: the arr IS NULL row must not reach it (#3178). - -statement -CREATE TABLE test_aj_guard(arr array, delims array, idx int) USING parquet - -statement -INSERT INTO test_aj_guard VALUES - (NULL, array(','), 0), - (array('a', 'b'), array(','), 1), - (array('a', NULL, 'b'), array(';'), 1) - --- the null-array row short-circuits before element_at(delims, 0) can throw -query -SELECT array_join(arr, element_at(delims, idx)) FROM test_aj_guard WHERE arr IS NULL - --- rows with a real index still join normally -query -SELECT array_join(arr, element_at(delims, idx)) FROM test_aj_guard WHERE arr IS NOT NULL - --- and the same shape with a null replacement -query -SELECT array_join(arr, element_at(delims, idx), 'X') FROM test_aj_guard WHERE arr IS NULL diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql index 2d9a77d8387..dee85de1d12 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql @@ -15,9 +15,9 @@ -- specific language governing permissions and limitations -- under the License. --- Regression coverage for #3178. Spark returns null whenever nullReplacement is null, even for +-- Regression coverage for #3178: Spark returns null whenever nullReplacement is null, even for -- an array with no nulls to replace, while array_to_string reads a null null_string as "omit --- nulls". +-- nulls". The replacement is a column so these take the guarded native path. statement CREATE TABLE test_aj_nullrep(arr array, delim string, nullrep string) USING parquet @@ -28,20 +28,15 @@ INSERT INTO test_aj_nullrep VALUES (array('a', 'b', 'c'), ',', NULL), (array(NULL, NULL), ',', NULL), (NULL, ',', NULL), - (array('a', NULL, 'c'), ',', 'X') + (array('a', NULL, 'c'), ',', 'X'), + (array('a', NULL, 'c'), ',', '') --- null replacement as a column, mixed with a non-null replacement row query SELECT array_join(arr, delim, nullrep) FROM test_aj_nullrep --- null replacement as a literal, array containing nulls query -SELECT array_join(array('a', NULL, 'b'), ',', cast(NULL as string)) +SELECT array_join(arr, ',', nullrep) FROM test_aj_nullrep --- null replacement as a literal, array containing no nulls at all -query -SELECT array_join(array('a', 'b'), ',', cast(NULL as string)) - --- a non-nullable literal replacement takes the unwrapped path +-- a non-nullable literal replacement takes the unguarded path query SELECT array_join(array('a', NULL, 'b'), ',', 'X') diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index 206706e68a7..984048af877 100644 --- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala @@ -25,7 +25,7 @@ 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.{ArrayContains, ArrayRemove} -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, If, Literal, MonotonicallyIncreasingID, Remainder} +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 @@ -33,7 +33,7 @@ import org.apache.spark.sql.types.{ArrayType, StringType} import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus} import org.apache.comet.DataTypeSupport.isComplexType -import org.apache.comet.serde.{CometArrayExcept, CometArrayJoin, CometArrayRemove, CometArrayReverse, CometFlatten, Compatible, Incompatible} +import org.apache.comet.serde.{CometArrayExcept, CometArrayJoin, CometArrayRemove, CometArrayReverse, CometFlatten, Compatible, ExprOuterClass, Incompatible} import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions} class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { @@ -536,71 +536,70 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp checkSparkAnswerAndOperator( sql( "SELECT array_join(array('hello', '-', 'world', cast(_2 as string)), ' ') from t1")) - // nullable, non-foldable delimiter and null replacement: the guarded shape + // column delimiter and nullable column replacement: the guarded native shape checkSparkAnswerAndOperator( - sql("SELECT array_join(array('a', cast(_2 as string), 'b'), cast(_6 as string), " + - "cast(_1 as string)) from t1")) + sql("SELECT array_join(array('a', cast(_2 as string), 'b'), _8, _8) from t1")) } } } } - // Result assertions cannot tell the native path from the codegen dispatcher, since an - // Incompatible verdict runs Spark's own doGenCode and matches. Pin the verdict itself. + // Result assertions cannot tell native from the dispatcher: an Incompatible verdict runs + // Spark's own doGenCode and matches. Pin the verdict itself. test("array_join support level pins the native path") { val nullableArray = AttributeReference("arr", ArrayType(StringType), nullable = true)() val nullableStr = AttributeReference("s", StringType, nullable = true)() + val delims = AttributeReference("delims", ArrayType(StringType), nullable = true)() + + // literals and column reads stay native + Seq( + ArrayJoin(nullableArray, Literal(","), None), + ArrayJoin(nullableArray, Literal(","), Some(Literal("X"))), + ArrayJoin(nullableArray, nullableStr, Some(nullableStr)), + // the array is unrestricted: it is evaluated on every path + ArrayJoin(ElementAt(delims, Literal(1)), Literal(","), None)).foreach { expr => + assert( + CometArrayJoin.getSupportLevel(expr).isInstanceOf[Compatible], + s"expected Compatible for $expr") + } - // literal delimiter, no replacement - assert( - CometArrayJoin - .getSupportLevel(ArrayJoin(nullableArray, Literal(","), None)) - .isInstanceOf[Compatible]) - // literal delimiter and literal replacement - assert( - CometArrayJoin - .getSupportLevel(ArrayJoin(nullableArray, Literal(","), Some(Literal("X")))) - .isInstanceOf[Compatible]) - // nullable column replacement still takes the native path, behind the guard - assert( - CometArrayJoin - .getSupportLevel(ArrayJoin(nullableArray, Literal(","), Some(nullableStr))) - .isInstanceOf[Compatible]) - - // A guarded non-deterministic argument would be evaluated twice. + // Anything that can throw or carry state goes to the dispatcher instead. + val throwingDelimiter = ElementAt(delims, Literal(0)) + val foldableThrowingDelimiter = ElementAt(CreateArray(Seq(Literal(","))), Literal(0)) val nonDeterministicReplacement = - If( - EqualTo(Remainder(MonotonicallyIncreasingID(), Literal(2L)), Literal(0L)), - Literal.create(null, StringType), - Literal("X")) - assert( - CometArrayJoin - .getSupportLevel( - ArrayJoin(nullableArray, Literal(","), Some(nonDeterministicReplacement))) - .isInstanceOf[Incompatible]) + Cast(MonotonicallyIncreasingID(), StringType) + Seq( + ArrayJoin(nullableArray, throwingDelimiter, None), + ArrayJoin(nullableArray, throwingDelimiter, Some(nullableStr)), + ArrayJoin(nullableArray, foldableThrowingDelimiter, None), + ArrayJoin(nullableArray, Literal(","), Some(nonDeterministicReplacement))).foreach { expr => + assert( + CometArrayJoin.getSupportLevel(expr).isInstanceOf[Incompatible], + s"expected Incompatible for $expr") + } } - test("array_join emits a null guard only where it is needed") { + test("array_join guards only a nullable replacement") { val nullableArray = AttributeReference("arr", ArrayType(StringType), nullable = true)() val nullableStr = AttributeReference("s", StringType, nullable = true)() val inputs = Seq(nullableArray, nullableStr) - def convert(expr: ArrayJoin) = CometArrayJoin.convert(expr, inputs, binding = false) + def convert(expr: ArrayJoin): Option[ExprOuterClass.Expr] = + CometArrayJoin.convert(expr, inputs, binding = false) - // All-foldable trailing arguments: no guard, so the plan is unchanged. - val unguarded = convert(ArrayJoin(nullableArray, Literal(","), Some(Literal("X")))) - assert(unguarded.isDefined) - assert(!unguarded.get.hasIf, "no guard should be emitted for a non-nullable replacement") + // No replacement, or a non-nullable one: unchanged plan. + val noReplacement = convert(ArrayJoin(nullableArray, Literal(","), None)) + assert(noReplacement.isDefined && !noReplacement.get.hasIf) + val literalReplacement = convert(ArrayJoin(nullableArray, Literal(","), Some(Literal("X")))) + assert(literalReplacement.isDefined && !literalReplacement.get.hasIf) - // Nullable replacement: Spark nullifies the row. + // A nullable replacement nullifies the row in Spark. val guarded = convert(ArrayJoin(nullableArray, Literal(","), Some(nullableStr))) - assert(guarded.isDefined) - assert(guarded.get.hasIf, "a nullable replacement must be guarded") + assert(guarded.isDefined && guarded.get.hasIf) - // Non-foldable delimiter: Spark never evaluates it when the array is null. - val arrayGuarded = convert(ArrayJoin(nullableArray, nullableStr, None)) - assert(arrayGuarded.isDefined) - assert(arrayGuarded.get.hasIf, "a non-foldable delimiter must put the array behind a guard") + // A nullable delimiter needs none: array_to_string already returns null for it. + val nullableDelimiter = convert(ArrayJoin(nullableArray, nullableStr, None)) + assert(nullableDelimiter.isDefined && !nullableDelimiter.get.hasIf) } test("arrays_overlap") { From 7401e0f602a8116812384e456361b73eeff70282 Mon Sep 17 00:00:00 2001 From: Viacheslav Inozemtsev Date: Thu, 3 Sep 2026 11:33:28 +0300 Subject: [PATCH 4/4] test: cover a literal NULL replacement on array_join's native path --- .../org/apache/comet/CometArrayExpressionSuite.scala | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index 984048af877..247de1bb1ef 100644 --- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala @@ -539,6 +539,11 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp // column delimiter and nullable column replacement: the guarded native shape checkSparkAnswerAndOperator( sql("SELECT array_join(array('a', cast(_2 as string), 'b'), _8, _8) from t1")) + // a literal NULL replacement folds to Literal(null, StringType), which is + // order-insensitive, so this takes the native path rather than the dispatcher. The + // sql-tests fixtures cannot reach this shape because they disable ConstantFolding. + checkSparkAnswerAndOperator( + sql("SELECT array_join(array('a', cast(_2 as string), 'b'), ',', NULL) from t1")) } } } @@ -556,6 +561,7 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp ArrayJoin(nullableArray, Literal(","), None), ArrayJoin(nullableArray, Literal(","), Some(Literal("X"))), ArrayJoin(nullableArray, nullableStr, Some(nullableStr)), + ArrayJoin(nullableArray, Literal(","), Some(Literal.create(null, StringType))), // the array is unrestricted: it is evaluated on every path ArrayJoin(ElementAt(delims, Literal(1)), Literal(","), None)).foreach { expr => assert( @@ -596,6 +602,9 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp // A nullable replacement nullifies the row in Spark. val guarded = convert(ArrayJoin(nullableArray, Literal(","), Some(nullableStr))) assert(guarded.isDefined && guarded.get.hasIf) + val literalNull = + convert(ArrayJoin(nullableArray, Literal(","), Some(Literal.create(null, StringType)))) + assert(literalNull.isDefined && literalNull.get.hasIf) // A nullable delimiter needs none: array_to_string already returns null for it. val nullableDelimiter = convert(ArrayJoin(nullableArray, nullableStr, None))