You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/20 19:29:56 UTC
(datafusion-comet) 01/01: fix: evaluate the array_append item on rows whose array is NULL (#7025)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7025-842a37b304d4af734fc275b8a8f8a7410080c193
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit d2c4901c7005a64360a29936125c2f77b9dea8d2
Author: KAZUYUKI TANIMURA <kt...@apple.com>
AuthorDate: Tue Oct 20 19:29:32 2026 +0000
fix: evaluate the array_append item on rows whose array is NULL (#7025)
* fix: array_append: ANSI item error is swallowed when the array is NUL
* address review comments
* 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 <packaging>jar|bundle</packaging>. A POM with
no <packaging> 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 <packaging>pom</packaging>
parent has no JAR by design and is kept.
* Revert "ci: purge POM-only Maven entries that declare no packaging"
This reverts commit dc74c67d228e15b2f910577e370876ea5c975c5a.
* address review comments
---
.../expression-audits/array_funcs.md | 1 +
.../spark-expr/src/conditional_funcs/case_when.rs | 86 ++++++++++++++
.../main/scala/org/apache/comet/serde/arrays.scala | 79 +++++++++++-
.../expressions/array/array_append_eager_eval.sql | 132 +++++++++++++++++++++
4 files changed, 295 insertions(+), 3 deletions(-)
diff --git a/docs/source/contributor-guide/expression-audits/array_funcs.md b/docs/source/contributor-guide/expression-audits/array_funcs.md
index 87ff575307..a2945aa95f 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)): 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 c [...]
## 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 02891e00f1..3dca671a51 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<dyn PhysicalExpr>,
+ Arc::clone(&null_else),
+ )],
+ Some(null_else),
+ &schema,
+ )
+ .unwrap();
+ create_case_when(
+ vec![(
+ Arc::new(IsNotNullExpr::new(c("arr"))) as Arc<dyn PhysicalExpr>,
+ c("arr"),
+ )],
+ Some(evaluate_item),
+ &schema,
+ )
+ .unwrap()
+ };
+ let batch = |arrays: Vec<Option<Vec<Option<i32>>>>, divisors: Vec<Option<i32>>| {
+ let len = arrays.len();
+ RecordBatch::try_new(
+ Arc::new(schema.clone()),
+ vec![
+ Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(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 812d792411..00018bae05 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,40 @@ 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). With a nullable array, `convert` reproduces that by also evaluating the
+ * item in the `ELSE` branch of its `CASE WHEN <array> 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 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
+ * 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.children.exists(!_.deterministic)) {
+ Unsupported(Some(nondeterministicReason))
+ } else {
+ Compatible()
+ }
+ }
+
override def convert(
expr: ArrayAppend,
inputs: Seq[Attribute],
@@ -83,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(
@@ -91,15 +132,47 @@ 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
+ // branch, so that it raises there as well, and still once per row. See
+ // `nondeterministicReason` for why this is restricted to deterministic operands.
+ val elseExprProto =
+ if (cannotThrow(expr.children(1))) {
+ 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 && nullLiteralProto.isDefined) {
+ 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 0000000000..93e9caa0b8
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_append_eager_eval.sql
@@ -0,0 +1,132 @@
+-- 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 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_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_class(DIVIDE_BY_ZERO)
+SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY<INT>)), 1 / (_1 - 1)) AS a
+FROM array_append_eager
+
+-- The item still raises on a row whose array is not NULL
+query expect_error_class(DIVIDE_BY_ZERO)
+SELECT _1, array_append(IF(_1 % 2 = 1, array(1), CAST(NULL AS ARRAY<INT>)), 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<INT>)), 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<INT>)), _1) AS a,
+ array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY<INT>)), 5) AS b
+FROM array_append_eager
+
+-- 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<INT>)), 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<INT>), 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_class(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<INT>) USING parquet
+
+statement
+INSERT INTO array_append_all_null SELECT /*+ COALESCE(1) */ * FROM VALUES (0, NULL), (1, NULL), (2, NULL), (3, NULL)
+
+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)
+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<INT>) USING parquet
+
+statement
+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_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 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<INT>>)), array(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<INT>>)), 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(1), CAST(NULL AS ARRAY<INT>)), _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<INT>)), 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
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org