fix: evaluate the array_append item on rows whose array is NULL

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 codegen. Its `CASE WHEN <array> 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
 

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

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

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