fix: preserve array_join evaluation semantics across Spark versions

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` 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)).
+- 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 other expressions may require codegen dispatch, 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)). Before Spark 4.2, nullable non-literal replacements stay with the enclosing Spark operator to preserve effective-nullability behavior ([SPARK-57200](https://issues.apache.org/jira/browse/SPARK-57200)). Compound nullable replacements always stay on Spark to avoid repeated evaluation, as does `NO_CODEGEN` execution to preserve interpreted argument order. The root serializer checks nested joins before parent codegen dispatch can hide them.
 
 ## array_max
 

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 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_join` | ✅ | Hybrid | Native for literal or column delimiter and replacement, except nullable non-literal replacements before Spark 4.2. Compound nullable replacements and NO_CODEGEN stay on Spark; other incompatible cases 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 |

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -870,6 +870,27 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       inputs: Seq[Attribute],
       binding: Boolean = true): Option[Expr] = {
 
+    // Some ArrayJoin shapes need the enclosing Spark operator, including when a parent would
+    // otherwise hide them inside codegen dispatch. Inspect independent roots once, before decimal
+    // promotion or recursive serialization. A worklist avoids rebuilding or recursing through
+    // deep expression trees.
+    var remaining = expr :: Nil
+    while (remaining.nonEmpty) {
+      val node = remaining.head
+      remaining = remaining.tail
+      node match {
+        case join: ArrayJoin =>
+          CometArrayJoin.operatorFallbackReason(join) match {
+            case Some(reason) =>
+              withFallbackReason(expr, reason)
+              return None
+            case None =>
+          }
+        case _ =>
+      }
+      node.children.reverseIterator.foreach(child => remaining = child :: remaining)
+    }
+
     val newExpr = DecimalPrecision.promote(expr)
     val result = exprToProtoInternal(newExpr, inputs, binding)
     if (!(newExpr eq expr)) {

spark/src/main/scala/org/apache/comet/serde/arrays.scala

@@ -28,7 +28,7 @@ import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
 import org.apache.comet.CometConf
-import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
+import org.apache.comet.CometSparkSessionExtensions.{isSpark42Plus, withFallbackReason}
 import org.apache.comet.DataTypeSupport.{deepNullable, isComplexType}
 import org.apache.comet.serde.QueryPlanSerde._
 import org.apache.comet.shims.{CometExprShim, CometTypeShim}
@@ -374,26 +374,70 @@ object CometArrayJoin
    * 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.
+   * observed for it; anything else is not admitted natively by default. `foldable` is not usable
+   * here: ConstantFolding leaves a throwing foldable expression unfolded in a conditional branch.
    */
   private def orderInsensitive(expr: Expression): Boolean = expr match {
     case _: Literal | _: Attribute | _: BoundReference => true
     case _ => false
   }
 
-  override def getIncompatibleReasons(): Seq[String] = Seq(collationReason, eagerEvalReason)
+  private val nullableReplacementReason =
+    "array_join with a nullable non-literal replacement requires the surrounding Spark " +
+      "operator before Spark 4.2 (https://issues.apache.org/jira/browse/SPARK-57200)"
+
+  private val repeatedReplacementReason =
+    "array_join with a compound nullable replacement requires Spark execution to avoid " +
+      "evaluating the replacement more than once"
+
+  private def hasNullableNonLiteralReplacement(expr: ArrayJoin): Boolean =
+    expr.nullReplacement.exists {
+      case _: Literal => false
+      case replacement => replacement.nullable
+    }
+
+  private def hasCompoundNullableReplacement(expr: ArrayJoin): Boolean =
+    expr.nullReplacement.exists(replacement =>
+      replacement.nullable && !orderInsensitive(replacement))
+
+  // Check before a parent can dispatch the whole subtree. Isolated codegen cannot reproduce
+  // nullability inferred by the enclosing project/filter, or Spark's interpreted evaluation order.
+  private[serde] def operatorFallbackReason(expr: ArrayJoin): Option[String] = {
+    if (SQLConf.get.getConf(SQLConf.CODEGEN_FACTORY_MODE).toString == "NO_CODEGEN") {
+      Some(
+        "array_join requires Spark interpreted evaluation when " +
+          "spark.sql.codegen.factoryMode=NO_CODEGEN")
+    } else if (hasCompoundNullableReplacement(expr)) {
+      // The null guard serializes the replacement twice. Dispatching it in isolation is also
+      // unsafe: a native parent can serialize that dispatched expression more than once.
+      Some(repeatedReplacementReason)
+    } else if (!isSpark42Plus && hasNullableNonLiteralReplacement(expr) &&
+      !CometConf.isExprAllowIncompat(getExprConfigName(expr))) {
+      Some(nullableReplacementReason)
+    } else {
+      None
+    }
+  }
+
+  override def getIncompatibleReasons(): Seq[String] =
+    Seq(collationReason, eagerEvalReason, nullableReplacementReason)
+
+  override def getUnsupportedReasons(): Seq[String] = Seq(repeatedReplacementReason)
 
   override def getSupportLevel(expr: ArrayJoin): SupportLevel = {
     // Spark 4.0 widens ArrayJoin's input to StringTypeWithCollation. Concatenation itself is
     // collation-independent, so the joined value is always correct; only the output string's
     // collation metadata is dropped (Comet columns are UTF8_BINARY). Report Incompatible rather
     // than Unsupported so the JVM codegen dispatcher (Spark's own doGenCode) keeps collated
     // array_join native and matching Spark, consistent with CometReverse's #2190 handling.
-    if (hasNonDefaultStringCollation(expr.array.dataType)) {
+    if (hasCompoundNullableReplacement(expr)) {
+      Unsupported(Some(repeatedReplacementReason))
+    } else if (hasNonDefaultStringCollation(expr.array.dataType)) {
       Incompatible(Some(collationReason))
     } else if (!(expr.delimiter +: expr.nullReplacement.toSeq).forall(orderInsensitive)) {
       Incompatible(Some(eagerEvalReason))
+    } else if (!isSpark42Plus && hasNullableNonLiteralReplacement(expr)) {
+      Incompatible(Some(nullableReplacementReason))
     } else {
       Compatible()
     }

spark/src/test/resources/sql-tests/expressions/array/array_join.sql

@@ -44,8 +44,8 @@ SELECT array_join(arr, ',') FROM test_array_join
 query
 SELECT array_join(arr, ',', 'NULL') FROM test_array_join
 
--- all three arguments as columns, including null delimiter and null replacement rows
-query
+-- A nullable column replacement stays on Spark before 4.2 (SPARK-57200).
+query spark_answer_only
 SELECT array_join(arr, delim, nullrep) FROM test_array_join
 
 -- column array with a column delimiter but a literal replacement

spark/src/test/resources/sql-tests/expressions/array/array_join_eager_eval_dispatch.sql

@@ -15,7 +15,7 @@
 -- 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,
+-- A delimiter or replacement that can throw or carry state requires Spark evaluation,
 -- 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.
 
@@ -36,13 +36,13 @@ 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
+query spark_answer_only
 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
+query spark_answer_only
 SELECT array_join(arr, element_at(delims, 1), nullrep) FROM test_aj_eager WHERE arr IS NOT NULL

spark/src/test/resources/sql-tests/expressions/array/array_join_interpreted.sql

@@ -0,0 +1,54 @@
+-- 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.
+
+-- ConfigMatrix: spark.comet.exec.scalaUDF.codegen.enabled=false,true
+-- Config: spark.comet.expression.ArrayJoin.allowIncompatible=false
+-- Config: spark.comet.expression.RegExpReplace.allowIncompatible=false
+-- Config: spark.sql.codegen.factoryMode=NO_CODEGEN
+-- Config: spark.sql.codegen.wholeStage=false
+
+-- ArrayJoin.eval evaluates the array before the replacement. Its generated code does the
+-- reverse, and before Spark 4.2 it also has different null results for some input schemas.
+-- Preserve interpreted execution, including when a parent could dispatch the whole subtree.
+statement
+CREATE TABLE test_aj_interpreted(arr array<string>, delim string, nr string, nested array<array<string>>) USING parquet
+
+statement
+INSERT INTO test_aj_interpreted VALUES
+  (array('a', NULL, 'b'), ',', 'X', array(array('a', NULL, 'b'))),
+  (array('a', NULL, 'b'), ',', NULL, array(array('a', NULL, 'b')))
+
+query expect_fallback(NO_CODEGEN)
+SELECT array_join(arr, delim, nr) FROM test_aj_interpreted WHERE arr IS NOT NULL AND delim IS NOT NULL
+
+query expect_fallback(NO_CODEGEN)
+SELECT length(array_join(arr, delim, nr)) FROM test_aj_interpreted WHERE arr IS NOT NULL AND delim IS NOT NULL
+
+query expect_fallback(NO_CODEGEN)
+SELECT regexp_replace(array_join(arr, delim, nr), delim, nr) FROM test_aj_interpreted WHERE arr IS NOT NULL AND delim IS NOT NULL
+
+query expect_fallback(NO_CODEGEN)
+SELECT array_join(element_at(nested, 1), ',', nr) FROM test_aj_interpreted
+
+-- Keep unrelated expressions native in this mode. The array_join queries above intentionally
+-- fall back, so their fallback assertions and the Scala routing test guard the error case.
+query
+SELECT element_at(nested, 1) FROM test_aj_interpreted
+
+-- Even a NULL replacement must not hide the array argument's error in interpreted execution.
+query expect_error(INVALID_INDEX_OF_ZERO)
+SELECT array_join(element_at(nested, 0), ',', nr) FROM test_aj_interpreted WHERE nr IS NULL

spark/src/test/resources/sql-tests/expressions/array/array_join_null_replacement.sql

@@ -17,7 +17,7 @@
 
 -- 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". The replacement is a column so these take the guarded native path.
+-- nulls". Nullable column replacements stay on Spark before 4.2 (SPARK-57200).
 
 statement
 CREATE TABLE test_aj_nullrep(arr array<string>, delim string, nullrep string) USING parquet
@@ -31,10 +31,10 @@ INSERT INTO test_aj_nullrep VALUES
   (array('a', NULL, 'c'), ',', 'X'),
   (array('a', NULL, 'c'), ',', '')
 
-query
+query spark_answer_only
 SELECT array_join(arr, delim, nullrep) FROM test_aj_nullrep
 
-query
+query spark_answer_only
 SELECT array_join(arr, ',', nullrep) FROM test_aj_nullrep
 
 -- a non-nullable literal replacement takes the unguarded path