From 7f41196bcba42d2a391cfd0fbd265951e072ed45 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Thu, 10 Sep 2026 18:57:37 +0000 Subject: [PATCH 1/2] fix: preserve array_join evaluation semantics across Spark versions --- .../expression-audits/array_funcs.md | 2 +- docs/source/user-guide/latest/expressions.md | 2 +- .../apache/comet/serde/QueryPlanSerde.scala | 21 ++ .../scala/org/apache/comet/serde/arrays.scala | 54 +++- .../expressions/array/array_join.sql | 4 +- .../array/array_join_eager_eval_dispatch.sql | 6 +- .../array/array_join_interpreted.sql | 54 ++++ .../array/array_join_null_replacement.sql | 6 +- .../comet/CometArrayExpressionSuite.scala | 234 +++++++++++++++++- 9 files changed, 355 insertions(+), 28 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_join_interpreted.sql diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index 5cdb222eb1a..097cb49d06a 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` 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 diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 231d80affbf..4e50cc3991e 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 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 | 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 1fa43f07de6..c0f155a0149 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -869,6 +869,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)) { 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 cca9f63f8bf..39013392d0a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/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,15 +374,55 @@ 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 @@ -390,10 +430,14 @@ object CometArrayJoin // 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() } 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 533b529fae2..2e7e9584948 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 @@ -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 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 index b934fa43f1c..42f404090f5 100644 --- 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 @@ -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,7 +36,7 @@ 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 @@ -44,5 +44,5 @@ 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 diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_join_interpreted.sql b/spark/src/test/resources/sql-tests/expressions/array/array_join_interpreted.sql new file mode 100644 index 00000000000..38ee8932ab3 --- /dev/null +++ b/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, delim string, nr string, nested array>) 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 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 dee85de1d12..e59ae31282d 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 @@ -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, 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 diff --git a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala index ad86dd15bc1..4ce3ac74cd6 100644 --- a/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometArrayExpressionSuite.scala @@ -25,15 +25,17 @@ 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, Cast, CreateArray, ElementAt, Literal, MonotonicallyIncreasingID} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, Concat, CreateArray, ElementAt, Expression, IsNotNull, Literal, MonotonicallyIncreasingID, Or, RegExpReplace} +import org.apache.spark.sql.comet.{CometFilterExec, CometProjectExec} +import org.apache.spark.sql.execution.{FilterExec, ProjectExec} 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.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus} +import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark42Plus} import org.apache.comet.DataTypeSupport.isComplexType -import org.apache.comet.serde.{CometArrayExcept, CometArrayJoin, CometArrayRemove, CometArrayReverse, CometFlatten, Compatible, ExprOuterClass, Incompatible} +import org.apache.comet.serde.{CometArrayExcept, CometArrayJoin, CometArrayRemove, CometArrayReverse, CometFlatten, Compatible, ExprOuterClass, Incompatible, QueryPlanSerde, Unsupported} import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions} class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { @@ -519,7 +521,7 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp } } - // No allowIncompatible opt-in: array_join runs natively by default now. + // No allowIncompatible opt-in: Spark-compatible array_join shapes run natively by default. test("array_join") { Seq(true, false).foreach { dictionaryEnabled => withTempDir { dir => @@ -536,9 +538,14 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp checkSparkAnswerAndOperator( sql( "SELECT array_join(array('hello', '-', 'world', cast(_2 as string)), ' ') from t1")) - // 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")) + // Before Spark 4.2, nullable non-literal replacements conservatively stay on Spark. + val columnReplacement = + sql("SELECT array_join(array('a', cast(_2 as string), 'b'), _8, _8) from t1") + if (isSpark42Plus) { + checkSparkAnswerAndOperator(columnReplacement) + } else { + checkSparkAnswer(columnReplacement) + } // 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. @@ -549,18 +556,17 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp } } - // 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") { + // Result assertions cannot distinguish the native, dispatcher, and Spark fallback paths. Pin + // the support verdict itself. + test("array_join support level pins execution routing") { 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 + // Literals and a column-free replacement stay native on every supported Spark version. Seq( 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 => @@ -569,7 +575,19 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp s"expected Compatible for $expr") } - // Anything that can throw or carry state goes to the dispatcher instead. + // Spark before 4.2 can infer tighter nullability around either a project or filter, so a + // nullable non-literal replacement conservatively stays on Spark regardless of the declared + // array and delimiter nullability. + val nullableColumnReplacement = + ArrayJoin(nullableArray, nullableStr, Some(nullableStr)) + val columnSupport = CometArrayJoin.getSupportLevel(nullableColumnReplacement) + if (isSpark42Plus) { + assert(columnSupport.isInstanceOf[Compatible]) + } else { + assert(columnSupport.isInstanceOf[Incompatible]) + } + + // Anything that can throw or carry state is not native by default. val throwingDelimiter = ElementAt(delims, Literal(0)) val foldableThrowingDelimiter = ElementAt(CreateArray(Seq(Literal(","))), Literal(0)) val nonDeterministicReplacement = @@ -583,6 +601,14 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp CometArrayJoin.getSupportLevel(expr).isInstanceOf[Incompatible], s"expected Incompatible for $expr") } + + // A compound nullable replacement is never eligible for the native path: its null guard + // would otherwise evaluate the expression twice, including under allowIncompatible=true. + val compoundNullableReplacement = Concat(Seq(nullableStr, Literal(""))) + assert(compoundNullableReplacement.nullable) + val unsupported = CometArrayJoin.getSupportLevel( + ArrayJoin(nullableArray, Literal(","), Some(compoundNullableReplacement))) + assert(unsupported.isInstanceOf[Unsupported]) } test("array_join guards only a nullable replacement") { @@ -611,6 +637,188 @@ class CometArrayExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelp assert(nullableDelimiter.isDefined && !nullableDelimiter.get.hasIf) } + test("array_join rejects unsafe subtrees before preparing deep expression trees") { + val array = AttributeReference("arr", ArrayType(StringType), nullable = true)() + val joined = ArrayJoin(array, Literal(","), Some(Literal("X"))) + val nested = (1 to 2048).foldLeft[Expression](IsNotNull(joined)) { case (left, _) => + Or(left, Literal(false)) + } + withSQLConf("spark.sql.codegen.factoryMode" -> "NO_CODEGEN") { + assert(QueryPlanSerde.exprToProto(nested, Seq(array)).isEmpty) + assert( + nested + .getTagValue(CometExplainInfo.FALLBACK_REASONS) + .exists(_.exists(_.contains("NO_CODEGEN")))) + } + } + + test("array_join keeps interpreted projections on Spark with native opt-in") { + withSQLConf( + "spark.sql.adaptive.enabled" -> "false", + "spark.sql.codegen.wholeStage" -> "false", + "spark.sql.codegen.factoryMode" -> "NO_CODEGEN") { + withTable("array_join_interpreted") { + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + sql( + "CREATE TABLE array_join_interpreted " + + "(arr ARRAY, delim STRING, nr STRING) USING parquet") + sql( + "INSERT INTO array_join_interpreted VALUES " + + "(array('a', NULL, 'b'), ',', 'X'), " + + "(array('a', NULL, 'b'), ',', NULL)") + } + Seq(false, true).foreach { dispatcherEnabled => + withSQLConf( + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> dispatcherEnabled.toString, + CometConf.getExprAllowIncompatConfigKey(classOf[ArrayJoin]) -> "true", + CometConf.getExprAllowIncompatConfigKey(classOf[RegExpReplace]) -> "false") { + // The parent's dispatcher must not bypass the interpreted-evaluation requirement, + // even when the user opts into ArrayJoin's otherwise incompatible native path. + Seq( + "array_join(arr, delim, nr)", + "regexp_replace(array_join(arr, delim, nr), delim, nr)").foreach { expression => + val (_, cometPlan) = checkSparkAnswer( + s"SELECT $expression FROM array_join_interpreted " + + "WHERE arr IS NOT NULL AND delim IS NOT NULL") + assert(collect(cometPlan) { case p: ProjectExec => p }.nonEmpty) + assert(collect(cometPlan) { case p: CometProjectExec => p }.isEmpty) + val explain = new ExtendedExplainInfo() + assert(explain.getFallbackReasons(cometPlan).exists(_.contains("NO_CODEGEN"))) + assert(!explain.getNativeExpressions(cometPlan).contains("array_join")) + assert(!explain.getCodegenDispatchExpressions(cometPlan).contains("array_join")) + assert(!explain.getCodegenDispatchExpressions(cometPlan).contains("regexp_replace")) + } + } + } + } + } + } + + test("array_join never guards a compound nullable replacement natively") { + withSQLConf( + "spark.sql.adaptive.enabled" -> "false", + "spark.sql.codegen.factoryMode" -> "CODEGEN_ONLY", + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true", + CometConf.getExprAllowIncompatConfigKey(classOf[ArrayJoin]) -> "true") { + withTable("array_join_compound_replacement") { + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + sql( + "CREATE TABLE array_join_compound_replacement " + + "(arr ARRAY, delim STRING, nr STRING) USING parquet") + sql( + "INSERT INTO array_join_compound_replacement VALUES " + + "(array('a', NULL, 'b'), ',', 'X'), " + + "(array('a', NULL, 'b'), ',', NULL)") + } + + // The explicit compatibility opt-in remains available for a simple nullable column. This + // shape does not tighten arr/delim, so Spark and the opted-in native path still agree. + val (_, simplePlan) = checkSparkAnswerAndOperator( + sql( + "SELECT array_join(arr, delim, nr) " + + "FROM array_join_compound_replacement"), + Seq(classOf[CometProjectExec])) + val explain = new ExtendedExplainInfo() + assert(explain.getNativeExpressions(simplePlan).contains("array_join")) + + Seq( + "array_join(arr, delim, concat(nr, ''))", + "coalesce(array_join(arr, delim, " + + "concat(cast(monotonically_increasing_id() AS string), nr)), 'fallback')") + .foreach { expression => + val query = s"SELECT $expression FROM array_join_compound_replacement" + val (_, cometPlan) = checkSparkAnswer(query) + assert(collect(cometPlan) { case p: ProjectExec => p }.nonEmpty) + assert(collect(cometPlan) { case p: CometProjectExec => p }.isEmpty) + assert(explain.getFallbackReasons(cometPlan).exists(_.contains("more than once"))) + assert(!explain.getCodegenDispatchExpressions(cometPlan).contains("array_join")) + assert(!explain.getNativeExpressions(cometPlan).contains("array_join")) + } + } + } + } + + test("array_join preserves pre-4.2 codegen behavior in projects and filters") { + withSQLConf( + "spark.sql.adaptive.enabled" -> "false", + "spark.sql.codegen.wholeStage" -> "true", + "spark.sql.codegen.factoryMode" -> "CODEGEN_ONLY") { + withTable("array_join_filtered") { + // A materialized source prevents ConvertToLocalRelation from using interpreted eval. + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + sql( + "CREATE TABLE array_join_filtered " + + "(arr ARRAY, delim STRING, nr STRING) USING parquet") + sql( + "INSERT INTO array_join_filtered VALUES " + + "(array('a', NULL, 'b'), ',', 'X'), " + + "(array('a', NULL, 'b'), ',', NULL), " + + "(NULL, ',', 'X'), " + + "(array('a', NULL, 'b'), NULL, 'X')") + } + + Seq(false, true).foreach { dispatcherEnabled => + withSQLConf( + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> dispatcherEnabled.toString, + CometConf.getExprAllowIncompatConfigKey(classOf[ArrayJoin]) -> "false", + CometConf.getExprAllowIncompatConfigKey(classOf[RegExpReplace]) -> "false") { + val query = "SELECT array_join(arr, delim, nr) FROM array_join_filtered " + + "WHERE arr IS NOT NULL AND delim IS NOT NULL" + val (_, cometPlan) = if (!isSpark42Plus) { + checkSparkAnswer(query) + } else { + checkSparkAnswerAndOperator(sql(query), Seq(classOf[CometProjectExec])) + } + + val explain = new ExtendedExplainInfo() + if (!isSpark42Plus) { + assert(collect(cometPlan) { case p: ProjectExec => p }.nonEmpty) + assert(collect(cometPlan) { case p: CometProjectExec => p }.isEmpty) + assert(explain.getFallbackReasons(cometPlan).exists(_.contains("SPARK-57200"))) + } else { + assert(explain.getNativeExpressions(cometPlan).contains("array_join")) + assert(!explain.getCodegenDispatchExpressions(cometPlan).contains("array_join")) + } + + // A parent dispatcher cannot reconstruct nullability inherited from the child plan, + // so the pre-4.2 scan must find a nested ArrayJoin before the parent hides its subtree. + val parentQuery = "SELECT regexp_replace(array_join(arr, delim, nr), delim, nr) " + + "FROM array_join_filtered WHERE arr IS NOT NULL AND delim IS NOT NULL" + if (isSpark42Plus && dispatcherEnabled) { + val (_, parentPlan) = + checkSparkAnswerAndOperator(sql(parentQuery), Seq(classOf[CometProjectExec])) + assert(explain.getCodegenDispatchExpressions(parentPlan).contains("regexp_replace")) + } else { + checkSparkAnswer(parentQuery) + } + + // Spark's FilterExec codegen tightens attribute nullability after the leading + // IS NOT NULL checks. An isolated dispatcher cannot reconstruct that state, so the + // complete filter stays on Spark before 4.2. + val filterQuery = "SELECT * FROM array_join_filtered " + + "WHERE arr IS NOT NULL AND delim IS NOT NULL " + + "AND array_join(arr, delim, nr) = 'a,X,b'" + val (_, filterPlan) = if (!isSpark42Plus) { + checkSparkAnswer(filterQuery) + } else { + checkSparkAnswerAndOperator(sql(filterQuery), Seq(classOf[CometFilterExec])) + } + if (!isSpark42Plus) { + assert(collect(filterPlan) { case f: FilterExec => f }.nonEmpty) + assert(collect(filterPlan) { case f: CometFilterExec => f }.isEmpty) + assert(explain.getFallbackReasons(filterPlan).exists(_.contains("SPARK-57200"))) + } else { + assert(collect(filterPlan) { case f: CometFilterExec => f }.nonEmpty) + assert(explain.getNativeExpressions(filterPlan).contains("array_join")) + } + } + } + } + } + } + test("arrays_overlap") { Seq(true, false).foreach { dictionaryEnabled => withTempDir { dir => From 12e3d85c694e4349769246881037f4f547cbde17 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Thu, 10 Sep 2026 21:38:04 +0000 Subject: [PATCH 2/2] test: distinguish native decimal coverage from array_join fallback --- .../comet/CometDecimalPromotionSuite.scala | 28 +++++++++++++++++-- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/CometDecimalPromotionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/CometDecimalPromotionSuite.scala index 8181628242f..9fd2b266ba4 100644 --- a/spark/src/test/scala/org/apache/spark/sql/comet/CometDecimalPromotionSuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/comet/CometDecimalPromotionSuite.scala @@ -22,6 +22,7 @@ package org.apache.spark.sql.comet import org.apache.spark.sql.CometTestBase import org.apache.spark.sql.catalyst.expressions.{Add, ArrayContains, AttributeReference, BitwiseNot, Cast, CreateArray, Divide, EvalMode, Multiply, NamedExpression} import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Partial, Sum} +import org.apache.spark.sql.execution.ProjectExec import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{DecimalType, IntegerType} @@ -189,9 +190,10 @@ class CometDecimalPromotionSuite extends CometTestBase { "array_compact(array($e))", "array_except(array($e), array($e))", "array_join(array(CAST($e AS STRING)), ',')", - // Spark's ArrayJoin codegen needs a nullable array or delimiter to clear isNull - // when the replacement is nullable. Use a column-based delimiter for this case. - "array_join(array('x', NULL), CAST(a AS STRING), CAST($e AS STRING))", + "array_join(array('x', NULL), CAST($e AS STRING), 'replacement')", + // Keep decimal arithmetic in the replacement's native serializer path. Coalesce + // makes it non-nullable, so ArrayJoin does not need a repeated-evaluation guard. + "array_join(array('x', NULL), ',', coalesce(CAST($e AS STRING), 'overflow'))", "slice(array($e), 1, 1)", "slice(array(1, 2), CAST($e AS INT), 2)", "slice(array(1, 2), 1, CAST($e AS INT))", @@ -222,6 +224,26 @@ class CometDecimalPromotionSuite extends CometTestBase { } } } + + // A nullable compound replacement deliberately keeps the project on Spark, even + // with native compatibility opt-in. Preserve result/error coverage for this shape + // separately from the native recursive-serialization cases above. + Seq("a * b", "c / d").foreach { arithmetic => + val query = "SELECT array_join(array('x', NULL), CAST(a AS STRING), " + + s"CAST($arithmetic AS STRING)) FROM decimal_overflow" + withClue(query) { + val df = sql(query) + val plan = df.queryExecution.executedPlan + assert(collect(plan) { case p: ProjectExec => p }.nonEmpty) + assert(collect(plan) { case p: CometProjectExec => p }.isEmpty) + if (ansi) { + val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df) + assert(sparkError.isDefined && cometError.isDefined) + } else { + checkSparkAnswerAndFallbackReason(df, "compound nullable replacement") + } + } + } } } }