From f23de9bffd4bf078345b66973210be7c8ed83ad8 Mon Sep 17 00:00:00 2001 From: Kazuyuki Tanimura Date: Thu, 15 Oct 2026 11:37:21 -0700 Subject: [PATCH 1/5] fix: array_append: ANSI item error is swallowed when the array is NUL --- .../expression-audits/array_funcs.md | 1 + .../scala/org/apache/comet/serde/arrays.scala | 59 +++++++++++++++++- .../array/array_append_eager_eval.sql | 61 +++++++++++++++++++ 3 files changed, 119 insertions(+), 2 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index af635270e74..fefcf9bf2d9 100644 --- a/docs/source/contributor-guide/expression-audits/array_funcs.md +++ b/docs/source/contributor-guide/expression-audits/array_funcs.md @@ -34,6 +34,7 @@ - Spark 3.5.8 (audited 2026-05-27): identical to 3.4.3. - Spark 4.0.1 (audited 2026-05-27): now `RuntimeReplaceable` and rewritten to `ArrayInsert(arr, Literal(-1), elem)`. `CometArrayAppend` is therefore unreachable; dispatch goes through `CometArrayInsert` (which carries its own `Incompatible` notes documented at the `array_insert` entry). - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1. +- Spark 3.4, 3.5 (2026-10-15, [#6086](https://github.com/apache/datafusion-comet/issues/6086)): `doGenCode` emits the item's code outside `nullSafeExec`, so Spark evaluates the item on every row, including rows whose array is `NULL`, and an item that raises under ANSI raises there. Comet's `CASE WHEN IS NOT NULL THEN array_append(...) ELSE null` guard skipped the item on those rows; the `ELSE` branch now evaluates it too. A nondeterministic item with a nullable array falls back to Spark, because the guard serializes the item twice. ## array_compact 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 e8778b335e9..1c068fe6a36 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -62,6 +62,35 @@ object CometArrayRemove object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBase { + /** + * Spark's generated code for `ArrayAppend` evaluates the item for every row and guards only the + * assignment of the result, so an item that raises under ANSI raises on a row whose array is + * NULL too (#6086). `convert` reproduces that by also evaluating the item in the `ELSE` branch + * of its `CASE WHEN IS NOT NULL` guard, which means it serializes the item twice. A + * nondeterministic item cannot survive that: each copy sees only a subset of the rows, so they + * advance its state independently and silently move values and NULLs. + */ + private val nondeterministicItemReason: String = + "array_append with a nullable array and a nondeterministic item: the native guard on the " + + "array evaluates the item on a subset of the rows, where Spark evaluates it on every row" + + /** + * Whether evaluating `expr` for a row Spark would not guard it on is unobservable: a literal or + * column read cannot throw. `foldable` is not usable here, see CometArrayJoin.orderInsensitive. + */ + private def cannotThrow(expr: Expression): Boolean = expr match { + case _: Literal | _: Attribute | _: BoundReference => true + case _ => false + } + + override def getSupportLevel(expr: ArrayAppend): SupportLevel = { + if (expr.left.nullable && !expr.right.deterministic) { + Unsupported(Some(nondeterministicItemReason)) + } else { + Compatible() + } + } + override def convert( expr: ArrayAppend, inputs: Seq[Attribute], @@ -94,12 +123,38 @@ object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBas val nullLiteralProto = exprToProtoInternal(Literal(null, elementType), Seq.empty, binding = true) - if (arrayAppendScalarExpr.isDefined && isNotNullExpr.isDefined && nullLiteralProto.isDefined) { + // DataFusion's CaseExpr evaluates the THEN branch only on the rows the guard selects, but + // Spark evaluates the item on every row. Evaluate it on the remaining rows in the ELSE + // branch, so that it raises there as well, and still once per row. See + // `nondeterministicItemReason` for why this is restricted to deterministic items. + val elseExprProto = + if (cannotThrow(expr.children(1)) || !expr.children.head.nullable) { + nullLiteralProto + } else { + for { + itemProto <- keyExprProto + nullLit <- nullLiteralProto + } yield { + val itemIsNotNull = ExprOuterClass.Expr + .newBuilder() + .setIsNotNull(ExprOuterClass.UnaryExpr.newBuilder().setChild(itemProto)) + .build() + val evaluateItem = ExprOuterClass.CaseWhen + .newBuilder() + .addWhen(itemIsNotNull) + .addThen(nullLit) + .setElseExpr(nullLit) + .build() + ExprOuterClass.Expr.newBuilder().setCaseWhen(evaluateItem).build() + } + } + + if (arrayAppendScalarExpr.isDefined && isNotNullExpr.isDefined && elseExprProto.isDefined) { val caseWhenExpr = ExprOuterClass.CaseWhen .newBuilder() .addWhen(isNotNullExpr.get) .addThen(arrayAppendScalarExpr.get) - .setElseExpr(nullLiteralProto.get) + .setElseExpr(elseExprProto.get) .build() Some( ExprOuterClass.Expr diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql b/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql new file mode 100644 index 00000000000..e01048caab4 --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql @@ -0,0 +1,61 @@ +-- 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. + +-- ArrayAppend is RuntimeReplaceable from Spark 4.0, so CometArrayAppend is only reachable here. +-- MaxSparkVersion: 3.5 + +-- Spark's generated code for array_append evaluates the item on every row and guards only the +-- assignment of the result, so an item that raises under ANSI raises on a row whose array is NULL +-- as well. https://github.com/apache/datafusion-comet/issues/6086 +-- Config: spark.sql.ansi.enabled=true + +statement +CREATE TABLE array_append_eager(_1 INT) USING parquet + +statement +INSERT INTO array_append_eager VALUES (0), (1), (2), (3) + +-- Control: the ANSI divide raises on its own, so the query below is not vacuous +query expect_error(DIVIDE_BY_ZERO) +SELECT _1, 1 / (_1 - 1) AS d FROM array_append_eager + +-- At _1 = 1 the array is NULL and the item divides by zero +query expect_error(DIVIDE_BY_ZERO) +SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), 1 / (_1 - 1)) AS a +FROM array_append_eager + +-- The item still raises on a row whose array is not NULL +query expect_error(DIVIDE_BY_ZERO) +SELECT _1, array_append(IF(_1 % 2 = 1, array(1), CAST(NULL AS ARRAY)), 1 / (_1 - 1)) AS a +FROM array_append_eager + +-- An item that does not raise on any row is unaffected, and stays native +query expect_native(array_append) +SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), 1 / (_1 + 1)) AS a +FROM array_append_eager + +-- A column or literal item cannot raise, so it needs no extra evaluation +query expect_native(array_append) +SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), _1) AS a, + array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), 5) AS b +FROM array_append_eager + +-- A nondeterministic item is declined when the array can be NULL: the native guard would +-- serialize it twice, and each copy would see only a subset of the rows. +query expect_fallback(nondeterministic item) +SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), CAST(rand(7L) * 3 AS INT)) AS a +FROM array_append_eager From 160a13ba0bfad09384f50e297accc3d726ad5f40 Mon Sep 17 00:00:00 2001 From: Kazuyuki Tanimura Date: Thu, 15 Oct 2026 12:46:28 -0700 Subject: [PATCH 2/5] address review comments --- .../expression-audits/array_funcs.md | 2 +- .../scala/org/apache/comet/serde/arrays.scala | 34 ++++++---- .../array/array_append_eager_eval.sql | 62 ++++++++++++++++++- 3 files changed, 83 insertions(+), 15 deletions(-) diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index fefcf9bf2d9..5ce807834bb 100644 --- a/docs/source/contributor-guide/expression-audits/array_funcs.md +++ b/docs/source/contributor-guide/expression-audits/array_funcs.md @@ -34,7 +34,7 @@ - Spark 3.5.8 (audited 2026-05-27): identical to 3.4.3. - Spark 4.0.1 (audited 2026-05-27): now `RuntimeReplaceable` and rewritten to `ArrayInsert(arr, Literal(-1), elem)`. `CometArrayAppend` is therefore unreachable; dispatch goes through `CometArrayInsert` (which carries its own `Incompatible` notes documented at the `array_insert` entry). - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1. -- Spark 3.4, 3.5 (2026-10-15, [#6086](https://github.com/apache/datafusion-comet/issues/6086)): `doGenCode` emits the item's code outside `nullSafeExec`, so Spark evaluates the item on every row, including rows whose array is `NULL`, and an item that raises under ANSI raises there. Comet's `CASE WHEN IS NOT NULL THEN array_append(...) ELSE null` guard skipped the item on those rows; the `ELSE` branch now evaluates it too. A nondeterministic item with a nullable array falls back to Spark, because the guard serializes the item twice. +- Spark 3.4, 3.5 (2026-10-15, [#6086](https://github.com/apache/datafusion-comet/issues/6086)): `doGenCode` emits the item's code outside `nullSafeExec`, so Spark evaluates the item on every row, including rows whose array is `NULL`, and an item that raises under ANSI raises there. Comet's `CASE WHEN IS NOT NULL THEN array_append(...) ELSE null` guard skipped the item on those rows; the `ELSE` branch now evaluates it too. A nondeterministic array or item with a nullable array falls back to Spark, because the guard serializes the operands more than once. A non-nullable array gets no guard at all, so each operand is evaluated once over the whole batch. ## array_compact 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 1c068fe6a36..1ed6e638717 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -65,14 +65,19 @@ object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBas /** * Spark's generated code for `ArrayAppend` evaluates the item for every row and guards only the * assignment of the result, so an item that raises under ANSI raises on a row whose array is - * NULL too (#6086). `convert` reproduces that by also evaluating the item in the `ELSE` branch - * of its `CASE WHEN IS NOT NULL` guard, which means it serializes the item twice. A - * nondeterministic item cannot survive that: each copy sees only a subset of the rows, so they - * advance its state independently and silently move values and NULLs. + * NULL too (#6086). With a nullable array, `convert` reproduces that by also evaluating the + * item in the `ELSE` branch of its `CASE WHEN IS NOT NULL` guard. The guard serializes + * the array twice and the `ELSE` serializes the item a second time, and a stateful operand + * cannot survive that: each copy sees only a subset of the rows, so the copies advance its + * state independently and silently move values and NULLs. Either operand is declined when it is + * nondeterministic. A non-nullable array needs no guard, so `convert` emits no `CASE` for it + * and evaluates each operand once over the whole batch, in order, whatever the operands are. */ - private val nondeterministicItemReason: String = - "array_append with a nullable array and a nondeterministic item: the native guard on the " + - "array evaluates the item on a subset of the rows, where Spark evaluates it on every row" + private val nondeterministicReason: String = + "array_append with a nullable array and a nondeterministic operand such as `rand()`: the " + + "native NULL guard evaluates the operands more than once, on subsets of the rows" + + override def getUnsupportedReasons(): Seq[String] = Seq(nondeterministicReason) /** * Whether evaluating `expr` for a row Spark would not guard it on is unobservable: a literal or @@ -84,8 +89,8 @@ object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBas } override def getSupportLevel(expr: ArrayAppend): SupportLevel = { - if (expr.left.nullable && !expr.right.deterministic) { - Unsupported(Some(nondeterministicItemReason)) + if (expr.left.nullable && expr.children.exists(!_.deterministic)) { + Unsupported(Some(nondeterministicReason)) } else { Compatible() } @@ -112,6 +117,13 @@ object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBas arrayExprProto, keyExprProto) + // A non-nullable array never takes the NULL branch, so the guard is dead weight. Without it + // each operand is evaluated once over the whole batch, which also keeps a nondeterministic + // operand in step. + if (!expr.children.head.nullable) { + return arrayAppendScalarExpr + } + // IS NOT NULL does not care about element nullability, so use the un-widened source and skip // serializing a redundant cast. val isNotNullExpr = createUnaryExpr( @@ -126,9 +138,9 @@ object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBas // DataFusion's CaseExpr evaluates the THEN branch only on the rows the guard selects, but // Spark evaluates the item on every row. Evaluate it on the remaining rows in the ELSE // branch, so that it raises there as well, and still once per row. See - // `nondeterministicItemReason` for why this is restricted to deterministic items. + // `nondeterministicReason` for why this is restricted to deterministic operands. val elseExprProto = - if (cannotThrow(expr.children(1)) || !expr.children.head.nullable) { + if (cannotThrow(expr.children(1))) { nullLiteralProto } else { for { diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql b/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql index e01048caab4..9a4d761646c 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql @@ -54,8 +54,64 @@ SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), _1) array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), 5) AS b FROM array_append_eager --- A nondeterministic item is declined when the array can be NULL: the native guard would --- serialize it twice, and each copy would see only a subset of the rows. -query expect_fallback(nondeterministic item) +-- A nondeterministic operand is declined when the array can be NULL: the native guard would +-- serialize it more than once, and each copy would see only a subset of the rows. This covers the +-- item and the array alike. +query expect_fallback(nondeterministic operand) SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), CAST(rand(7L) * 3 AS INT)) AS a FROM array_append_eager + +query expect_fallback(nondeterministic operand) +SELECT _1, array_append(IF(rand(7L) < 2, CAST(NULL AS ARRAY), array(1)), _1) AS a +FROM array_append_eager + +-- A non-nullable array needs no guard, so `convert` emits none and the item is evaluated once over +-- the whole batch. A nondeterministic item therefore stays native and in step with Spark. +query expect_native(array_append) +SELECT _1, array_append(array(1, 2), CAST(rand(7L) * 3 AS INT)) AS a +FROM array_append_eager + +-- A non-nullable array still raises for a throwing item +query expect_error(DIVIDE_BY_ZERO) +SELECT _1, array_append(array(1, 2), 1 / (_1 - 1)) AS a +FROM array_append_eager + +-- Every array NULL: the batch only ever takes the ELSE branch, which must still evaluate the item +statement +CREATE TABLE array_append_all_null(_1 INT, arr ARRAY) USING parquet + +statement +INSERT INTO array_append_all_null VALUES (0, NULL), (1, NULL), (2, NULL), (3, NULL) + +query expect_error(DIVIDE_BY_ZERO) +SELECT _1, array_append(arr, 1 / (_1 - 1)) AS a FROM array_append_all_null + +query expect_native(array_append) +SELECT _1, array_append(arr, 1 / (_1 + 1)) AS a FROM array_append_all_null + +-- No array NULL: the batch only ever takes the THEN branch +statement +CREATE TABLE array_append_no_null(_1 INT, arr ARRAY) USING parquet + +statement +INSERT INTO array_append_no_null VALUES (0, array(1)), (1, array(2)), (2, array()), (3, array(4, 5)) + +query expect_error(DIVIDE_BY_ZERO) +SELECT _1, array_append(arr, 1 / (_1 - 1)) AS a FROM array_append_no_null + +query expect_native(array_append) +SELECT _1, array_append(arr, 1 / (_1 + 1)) AS a FROM array_append_no_null + +-- Complex element types: the item is widened in lockstep with the array, and the widened item is +-- what the ELSE branch evaluates. The array item divides by zero only where the array is NULL. +query expect_error(DIVIDE_BY_ZERO) +SELECT _1, array_append(IF(_1 % 2 = 0, array(array(1)), CAST(NULL AS ARRAY>)), array(1 / (_1 - 1))) AS a +FROM array_append_eager + +query expect_error(DIVIDE_BY_ZERO) +SELECT _1, array_append(IF(_1 % 2 = 0, array(map(1, 2)), CAST(NULL AS ARRAY>)), map(1, 1 / (_1 - 1))) AS a +FROM array_append_eager + +query expect_native(array_append) +SELECT _1, array_append(IF(_1 % 2 = 0, array(array(1)), CAST(NULL AS ARRAY>)), array(1 / (_1 + 1))) AS a +FROM array_append_eager From dc74c67d228e15b2f910577e370876ea5c975c5a Mon Sep 17 00:00:00 2001 From: Kazuyuki Tanimura Date: Thu, 15 Oct 2026 14:30:50 -0700 Subject: [PATCH 3/5] ci: purge POM-only Maven entries that declare no packaging The "Purge partial Maven cache entries" step only deleted a jar-less POM when it carried an explicit jar|bundle. A POM with no defaults to jar, and Arrow's POMs, among others, declare none, so a POM-only entry for one of them survived the purge. Coursier then treated the artifact as found locally and dropped the missing JAR, and sbt failed compiling sql-api with "object memory is not a member of package org.apache.arrow" in the Spark 3.5 "Pre-compile Spark Test classes" step. Read the packaging and default it to jar, as dev/local-ci.sh's purge_partial_poms already does. A jar-less pom parent has no JAR by design and is kept. --- .github/actions/setup-spark-builder/action.yaml | 8 +++++++- dev/local-ci.sh | 6 +++--- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/.github/actions/setup-spark-builder/action.yaml b/.github/actions/setup-spark-builder/action.yaml index 40f3a1e4a8e..c5f5b5f6418 100644 --- a/.github/actions/setup-spark-builder/action.yaml +++ b/.github/actions/setup-spark-builder/action.yaml @@ -89,12 +89,18 @@ runs: # declares the artifact "found locally", and fails on the missing JAR # without falling back to Maven Central. Delete those partial entries # so sbt re-fetches the full artifact remotely. + # + # A POM that declares no defaults to jar, so it counts as + # partial too. Arrow's POMs, org.antlr:antlr4's and many others declare + # none. A pom parent has no JAR by design, so it + # is kept. Keep this in step with purge_partial_poms in dev/local-ci.sh. for repo in "$HOME/.m2/repository" /root/.m2/repository; do [ -d "$repo" ] || continue find "$repo" -name '*.pom' | while read -r pom; do jar="${pom%.pom}.jar" [ -f "$jar" ] && continue - grep -q 'jar\|bundle' "$pom" 2>/dev/null || continue + packaging="$(sed -n 's:.*\(.*\).*:\1:p' "$pom" | head -1)" + case "${packaging:-jar}" in jar | bundle) ;; *) continue ;; esac rm -f "$pom" "${pom}.sha1" "${pom%.pom}.pom.lastUpdated" \ "$(dirname "$pom")/_remote.repositories" done diff --git a/dev/local-ci.sh b/dev/local-ci.sh index 853108aeef8..73ba076e32d 100755 --- a/dev/local-ci.sh +++ b/dev/local-ci.sh @@ -190,9 +190,9 @@ purge_parquet() { rm -rf "$dir" } -# setup-spark-builder greps for an explicit jar|bundle and -# so misses a POM declaring none, which Maven defaults to jar. org.antlr:antlr4 -# is one of those. A pom parent has no JAR by design. +# Mirrors the purge in setup-spark-builder. A POM declaring no is +# purged too, since Maven defaults it to jar. org.antlr:antlr4 and Arrow's are +# among those. A pom parent has no JAR by design. purge_partial_poms() { repo="$(maven_repo)" [ -d "$repo" ] || return 0 From 1575a31e59d4590dd42477e686579fc5932b3ad1 Mon Sep 17 00:00:00 2001 From: Kazuyuki Tanimura Date: Mon, 19 Oct 2026 09:00:29 -0700 Subject: [PATCH 4/5] Revert "ci: purge POM-only Maven entries that declare no packaging" This reverts commit dc74c67d228e15b2f910577e370876ea5c975c5a. --- .github/actions/setup-spark-builder/action.yaml | 8 +------- dev/local-ci.sh | 6 +++--- 2 files changed, 4 insertions(+), 10 deletions(-) diff --git a/.github/actions/setup-spark-builder/action.yaml b/.github/actions/setup-spark-builder/action.yaml index c5f5b5f6418..40f3a1e4a8e 100644 --- a/.github/actions/setup-spark-builder/action.yaml +++ b/.github/actions/setup-spark-builder/action.yaml @@ -89,18 +89,12 @@ runs: # declares the artifact "found locally", and fails on the missing JAR # without falling back to Maven Central. Delete those partial entries # so sbt re-fetches the full artifact remotely. - # - # A POM that declares no defaults to jar, so it counts as - # partial too. Arrow's POMs, org.antlr:antlr4's and many others declare - # none. A pom parent has no JAR by design, so it - # is kept. Keep this in step with purge_partial_poms in dev/local-ci.sh. for repo in "$HOME/.m2/repository" /root/.m2/repository; do [ -d "$repo" ] || continue find "$repo" -name '*.pom' | while read -r pom; do jar="${pom%.pom}.jar" [ -f "$jar" ] && continue - packaging="$(sed -n 's:.*\(.*\).*:\1:p' "$pom" | head -1)" - case "${packaging:-jar}" in jar | bundle) ;; *) continue ;; esac + grep -q 'jar\|bundle' "$pom" 2>/dev/null || continue rm -f "$pom" "${pom}.sha1" "${pom%.pom}.pom.lastUpdated" \ "$(dirname "$pom")/_remote.repositories" done diff --git a/dev/local-ci.sh b/dev/local-ci.sh index 73ba076e32d..853108aeef8 100755 --- a/dev/local-ci.sh +++ b/dev/local-ci.sh @@ -190,9 +190,9 @@ purge_parquet() { rm -rf "$dir" } -# Mirrors the purge in setup-spark-builder. A POM declaring no is -# purged too, since Maven defaults it to jar. org.antlr:antlr4 and Arrow's are -# among those. A pom parent has no JAR by design. +# setup-spark-builder greps for an explicit jar|bundle and +# so misses a POM declaring none, which Maven defaults to jar. org.antlr:antlr4 +# is one of those. A pom parent has no JAR by design. purge_partial_poms() { repo="$(maven_repo)" [ -d "$repo" ] || return 0 From a3a935b7bfaa4d7a4d7edde9ce813a8b1c066398 Mon Sep 17 00:00:00 2001 From: Kazuyuki Tanimura Date: Mon, 19 Oct 2026 09:19:21 -0700 Subject: [PATCH 5/5] address review comments --- .../expression-audits/array_funcs.md | 2 +- .../src/conditional_funcs/case_when.rs | 86 +++++++++++++++++++ .../scala/org/apache/comet/serde/arrays.scala | 8 +- .../array/array_append_eager_eval.sql | 45 ++++++---- 4 files changed, 124 insertions(+), 17 deletions(-) diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md index 5ce807834bb..9a21d44653c 100644 --- a/docs/source/contributor-guide/expression-audits/array_funcs.md +++ b/docs/source/contributor-guide/expression-audits/array_funcs.md @@ -34,7 +34,7 @@ - Spark 3.5.8 (audited 2026-05-27): identical to 3.4.3. - Spark 4.0.1 (audited 2026-05-27): now `RuntimeReplaceable` and rewritten to `ArrayInsert(arr, Literal(-1), elem)`. `CometArrayAppend` is therefore unreachable; dispatch goes through `CometArrayInsert` (which carries its own `Incompatible` notes documented at the `array_insert` entry). - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1. -- Spark 3.4, 3.5 (2026-10-15, [#6086](https://github.com/apache/datafusion-comet/issues/6086)): `doGenCode` emits the item's code outside `nullSafeExec`, so Spark evaluates the item on every row, including rows whose array is `NULL`, and an item that raises under ANSI raises there. Comet's `CASE WHEN IS NOT NULL THEN array_append(...) ELSE null` guard skipped the item on those rows; the `ELSE` branch now evaluates it too. A nondeterministic array or item with a nullable array falls back to Spark, because the guard serializes the operands more than once. A non-nullable array gets no guard at all, so each operand is evaluated once over the whole batch. +- Spark 3.4, 3.5 (2026-10-15, [#6086](https://github.com/apache/datafusion-comet/issues/6086)): Spark disagrees with itself here. Interpreted `ArrayAppend.eval` returns NULL for a NULL array without evaluating the item, but `doGenCode` emits the item's code outside `nullSafeExec`, so generated code evaluates the item on every row, including rows whose array is `NULL`, and an item that raises under ANSI raises there. Comet follows the generated code, which is what runs under whole-stage codegen. Its `CASE WHEN IS NOT NULL THEN array_append(...) ELSE null` guard skipped the item on those rows; the `ELSE` branch now evaluates it too, when it is not a literal or a column read, and returns a NULL typed as the list. A nondeterministic array or item with a nullable array falls back to Spark, because the guard serializes the operands more than once. A non-nullable array gets no guard at all, so each operand is evaluated once over the whole batch. ## array_compact diff --git a/native/spark-expr/src/conditional_funcs/case_when.rs b/native/spark-expr/src/conditional_funcs/case_when.rs index 02891e00f15..3dca671a51f 100644 --- a/native/spark-expr/src/conditional_funcs/case_when.rs +++ b/native/spark-expr/src/conditional_funcs/case_when.rs @@ -1145,6 +1145,92 @@ mod tests { Arc::new(BinaryExpr::new(left, op, right)) } + /// The plan `CometArrayAppend` serializes for a nullable array, with the array standing in for + /// the appended list: THEN the list, ELSE an evaluation of the item that returns NULL. Spark's + /// generated code evaluates the item for every row, so an item that fails must fail on a row + /// whose array is NULL (#6086). Both levels of ELSE are NULL of the list type THEN returns: an + /// element-typed NULL made the inner CASE return Int32 where the outer CASE returns lists, and + /// the two could not be merged. + #[test] + fn evaluating_the_item_for_null_arrays_keeps_the_list_type() { + use arrow::array::ListArray; + use arrow::datatypes::Int32Type; + + let list_type = DataType::List(Arc::new(Field::new("item", DataType::Int32, true))); + let schema = Schema::new(vec![ + Field::new("arr", list_type.clone(), true), + Field::new("a", DataType::Int32, true), + Field::new("b", DataType::Int32, true), + ]); + let c = |name: &str| col(name, &schema).unwrap(); + // CASE WHEN arr IS NOT NULL THEN arr + // ELSE CASE WHEN a / b IS NOT NULL THEN NULL ELSE NULL END END + let plan = |null_else: ScalarValue| { + let null_else = lit(null_else); + let evaluate_item = create_case_when( + vec![( + Arc::new(IsNotNullExpr::new(binary(c("a"), Operator::Divide, c("b")))) + as Arc, + Arc::clone(&null_else), + )], + Some(null_else), + &schema, + ) + .unwrap(); + create_case_when( + vec![( + Arc::new(IsNotNullExpr::new(c("arr"))) as Arc, + c("arr"), + )], + Some(evaluate_item), + &schema, + ) + .unwrap() + }; + let batch = |arrays: Vec>>>, divisors: Vec>| { + let len = arrays.len(); + RecordBatch::try_new( + Arc::new(schema.clone()), + vec![ + Arc::new(ListArray::from_iter_primitive::(arrays)), + Arc::new(Int32Array::from(vec![Some(6); len])), + Arc::new(Int32Array::from(divisors)), + ], + ) + .unwrap() + }; + let expr = plan(ScalarValue::try_new_null(&list_type).unwrap()); + + // Some arrays are NULL and the item divides without failing: the lists come back as they + // were, and the NULLs stay NULL, with the list type + let arrays = vec![Some(vec![Some(1)]), None, Some(vec![Some(2)]), None]; + let input = batch(arrays, vec![Some(2), Some(3), Some(0), Some(4)]); + let result = expr.evaluate(&input).unwrap().into_array(4).unwrap(); + assert_eq!(result.data_type(), &list_type); + assert_eq!(&result, input.column(0)); + + // Every array is NULL: every row chooses ELSE, which still has to produce lists + let input = batch(vec![None, None], vec![Some(2), Some(3)]); + let result = expr.evaluate(&input).unwrap().into_array(2).unwrap(); + assert_eq!(result.data_type(), &list_type); + assert_eq!(result.null_count(), 2); + + // No array is NULL: ELSE is never evaluated, so a zero divisor goes unnoticed + let input = batch( + vec![Some(vec![Some(1)]), Some(vec![])], + vec![Some(0), Some(0)], + ); + let result = expr.evaluate(&input).unwrap().into_array(2).unwrap(); + assert_eq!(&result, input.column(0)); + + // The item fails on a row whose array is NULL, as in Spark + let arrays = vec![Some(vec![Some(1)]), None]; + let err = expr + .evaluate(&batch(arrays, vec![Some(2), Some(0)])) + .unwrap_err(); + assert!(err.to_string().contains("Divide by zero"), "{err}"); + } + #[test] fn branch_that_can_fail_is_evaluated_lazily() { // CASE WHEN b != 0 THEN a / b ELSE 0 END: integer division fails for b = 0, which the 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 1ed6e638717..1934d8e118f 100644 --- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala +++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala @@ -132,8 +132,14 @@ object CometArrayAppend extends CometExpressionSerde[ArrayAppend] with ArraysBas binding, (builder, unaryExpr) => builder.setIsNotNull(unaryExpr)) + // The NULL a row with a NULL array returns. It has to be typed as the list that THEN returns, + // not as an element: typed as an element, the inner CASE below returns elements, and the outer + // CASE cannot merge those with the lists of its other branch. val nullLiteralProto = - exprToProtoInternal(Literal(null, elementType), Seq.empty, binding = true) + exprToProtoInternal( + Literal(null, ArrayType(elementType, containsNull = true)), + Seq.empty, + binding = true) // DataFusion's CaseExpr evaluates the THEN branch only on the rows the guard selects, but // Spark evaluates the item on every row. Evaluate it on the remaining rows in the ELSE diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql b/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql index 9a4d761646c..93e9caa0b8c 100644 --- a/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql +++ b/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql @@ -27,19 +27,19 @@ statement CREATE TABLE array_append_eager(_1 INT) USING parquet statement -INSERT INTO array_append_eager VALUES (0), (1), (2), (3) +INSERT INTO array_append_eager SELECT /*+ COALESCE(1) */ * FROM VALUES (0), (1), (2), (3) -- Control: the ANSI divide raises on its own, so the query below is not vacuous -query expect_error(DIVIDE_BY_ZERO) +query expect_error_class(DIVIDE_BY_ZERO) SELECT _1, 1 / (_1 - 1) AS d FROM array_append_eager -- At _1 = 1 the array is NULL and the item divides by zero -query expect_error(DIVIDE_BY_ZERO) +query expect_error_class(DIVIDE_BY_ZERO) SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), 1 / (_1 - 1)) AS a FROM array_append_eager -- The item still raises on a row whose array is not NULL -query expect_error(DIVIDE_BY_ZERO) +query expect_error_class(DIVIDE_BY_ZERO) SELECT _1, array_append(IF(_1 % 2 = 1, array(1), CAST(NULL AS ARRAY)), 1 / (_1 - 1)) AS a FROM array_append_eager @@ -72,7 +72,7 @@ SELECT _1, array_append(array(1, 2), CAST(rand(7L) * 3 AS INT)) AS a FROM array_append_eager -- A non-nullable array still raises for a throwing item -query expect_error(DIVIDE_BY_ZERO) +query expect_error_class(DIVIDE_BY_ZERO) SELECT _1, array_append(array(1, 2), 1 / (_1 - 1)) AS a FROM array_append_eager @@ -81,9 +81,9 @@ statement CREATE TABLE array_append_all_null(_1 INT, arr ARRAY) USING parquet statement -INSERT INTO array_append_all_null VALUES (0, NULL), (1, NULL), (2, NULL), (3, NULL) +INSERT INTO array_append_all_null SELECT /*+ COALESCE(1) */ * FROM VALUES (0, NULL), (1, NULL), (2, NULL), (3, NULL) -query expect_error(DIVIDE_BY_ZERO) +query expect_error_class(DIVIDE_BY_ZERO) SELECT _1, array_append(arr, 1 / (_1 - 1)) AS a FROM array_append_all_null query expect_native(array_append) @@ -94,24 +94,39 @@ statement CREATE TABLE array_append_no_null(_1 INT, arr ARRAY) USING parquet statement -INSERT INTO array_append_no_null VALUES (0, array(1)), (1, array(2)), (2, array()), (3, array(4, 5)) +INSERT INTO array_append_no_null SELECT /*+ COALESCE(1) */ * FROM VALUES (0, array(1)), (1, array(2)), (2, array()), (3, array(4, 5)) -query expect_error(DIVIDE_BY_ZERO) +query expect_error_class(DIVIDE_BY_ZERO) SELECT _1, array_append(arr, 1 / (_1 - 1)) AS a FROM array_append_no_null query expect_native(array_append) SELECT _1, array_append(arr, 1 / (_1 + 1)) AS a FROM array_append_no_null --- Complex element types: the item is widened in lockstep with the array, and the widened item is --- what the ELSE branch evaluates. The array item divides by zero only where the array is NULL. -query expect_error(DIVIDE_BY_ZERO) +-- Complex element types: the item is itself an array, and the ELSE branch's NULL has to be typed as +-- the list of arrays that THEN returns. The item divides by zero only where the array is NULL. +query expect_error_class(DIVIDE_BY_ZERO) SELECT _1, array_append(IF(_1 % 2 = 0, array(array(1)), CAST(NULL AS ARRAY>)), array(1 / (_1 - 1))) AS a FROM array_append_eager -query expect_error(DIVIDE_BY_ZERO) -SELECT _1, array_append(IF(_1 % 2 = 0, array(map(1, 2)), CAST(NULL AS ARRAY>)), map(1, 1 / (_1 - 1))) AS a +query expect_native(array_append) +SELECT _1, array_append(IF(_1 % 2 = 0, array(array(1)), CAST(NULL AS ARRAY>)), array(1 / (_1 + 1))) AS a FROM array_append_eager +-- Without ANSI nothing raises, and the default on Spark 3.4 and 3.5 is without ANSI. A computed +-- item still goes through the extra evaluation for rows whose array is NULL, whose NULL has to be +-- typed as the list the other branch returns, or the two cannot be merged. The division by zero +-- returns NULL, and the second query has it land on a row whose array is not NULL, so [1, NULL] is +-- visible in the result. +statement +SET spark.sql.ansi.enabled=false + query expect_native(array_append) -SELECT _1, array_append(IF(_1 % 2 = 0, array(array(1)), CAST(NULL AS ARRAY>)), array(1 / (_1 + 1))) AS a +SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY)), _1 + 1) AS a FROM array_append_eager + +query expect_native(array_append) +SELECT _1, array_append(IF(_1 % 2 = 1, array(1), CAST(NULL AS ARRAY)), 1 / (_1 - 1)) AS a +FROM array_append_eager + +query expect_native(array_append) +SELECT _1, array_append(arr, _1 + 1) AS a FROM array_append_all_null