fix: send a skipped shared subexpression to Spark where Spark's sharing depends on the plan

docs/source/contributor-guide/expression-audits/array_funcs.md

@@ -35,6 +35,7 @@
 - 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.
+- Spark 3.4, 3.5 (2026-10-21, [#5867](https://github.com/apache/datafusion-comet/pull/5867)): a nondeterministic operand over a nullable array now runs through the JVM codegen dispatcher instead of falling back to Spark, unless the result holds a calendar interval or an argument the dispatched tree can skip holds a subexpression the operator shares; those still fall back. A non-nullable array stays native whatever its operands.
 
 ## array_compact
 

docs/source/contributor-guide/sql-file-tests.md

@@ -479,8 +479,9 @@ common ones include:
 - **Signed zero** -- Spark parses a bare `-0.0` as `decimal(1,1)`, which has no signed
   zero, so coercion to `float`/`double` yields `+0.0`. `CAST(-0.0 AS DOUBLE)` and
   `CAST(-0.0 AS FLOAT)` have the same problem because the cast source is still the
-  decimal literal. Use `double('-0.0')` or `float('-0.0')` (equivalently
-  `CAST('-0.0' AS DOUBLE)`). Spark's array comparator also treats `+0.0` and `-0.0` as
+  decimal literal. Use the double literal `-0.0D`, `CAST(-0.0D AS FLOAT)` for a float,
+  or `double('-0.0')` / `float('-0.0')` (equivalently `CAST('-0.0' AS DOUBLE)`), all of
+  which keep the sign. Spark's array comparator also treats `+0.0` and `-0.0` as
   equal, so `sort_array(...)` is not a unique projection when both signs are present
   (the SQL test comparator distinguishes the bits). Prefer a sign-aware form such as
   `sort_array(transform(arr, x -> cast(x AS string)))`. A `query tolerance=...` check

docs/source/user-guide/latest/expressions.md

@@ -154,7 +154,7 @@ The tables below list every Spark built-in expression with its current status.
 | Function | Status | Implementation | Notes |
 | --- | --- | --- | --- |
 | `array` | ✅ | Native |  |
-| `array_append` | ✅ | Native |  |
+| `array_append` | ✅ | Hybrid | A nondeterministic operand over a nullable array routes through the JVM codegen dispatcher |
 | `array_compact` | ✅ | — |  |
 | `array_contains` | ✅ | Hybrid | Flat float/double element arrays use a native kernel with Spark's equality; nested float elements, non-default string collations, and a value that is not a literal or column over a nullable array, route through the JVM codegen dispatcher |
 | `array_distinct` | ✅ | Native | Floating-point elements fall back on Spark versions before 4.0.5, 4.1.4, and 4.2; signed-zero results may differ with native opt-in ([details](compatibility/floating-point.md)). Non-default string collations fall back |
@@ -170,8 +170,8 @@ The tables below list every Spark built-in expression with its current status.
 | `array_repeat` | ✅ | Native |  |
 | `array_union` | ✅ | Native | Floating-point elements fall back on Spark versions before 4.0.5, 4.1.4, and 4.2; signed-zero results may differ with native opt-in ([details](compatibility/floating-point.md)). Non-default string collations fall back |
 | `arrays_overlap` | ✅ | Hybrid | Non-default string collations use the JVM codegen dispatcher |
-| `arrays_zip` | ✅ | Native |  |
-| `element_at` | ✅ | Native |  |
+| `arrays_zip` | ✅ | Hybrid |  |
+| `element_at` | ✅ | Hybrid |  |
 | `flatten` | ✅ | Native | Binary/struct/map elements fall back |
 | `get` | ✅ | — |  |
 | `sequence` | ✅ | Hybrid | Integral types run natively; date/timestamp sequences use codegen dispatch |
@@ -205,10 +205,10 @@ The tables below list every Spark built-in expression with its current status.
 | Function | Status | Implementation | Notes |
 | --- | --- | --- | --- |
 | `array_size` | ✅ | — |  |
-| `cardinality` | ✅ | Native |  |
+| `cardinality` | ✅ | Hybrid |  |
 | `concat` | ✅ | Hybrid | Binary/array children and non-UTF8_BINARY collations route through the JVM codegen dispatcher |
 | `reverse` | ✅ | Hybrid | Binary input (Spark 4.2), arrays with binary, struct, or map elements, and collated strings, route through the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) |
-| `size` | ✅ | Native |  |
+| `size` | ✅ | Hybrid |  |
 
 ---
 
@@ -404,12 +404,12 @@ to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enab
 
 | Function | Status | Implementation | Notes |
 | --- | --- | --- | --- |
-| `element_at` | ✅ | Native |  |
+| `element_at` | ✅ | Hybrid |  |
 | `map` | ✅ | Codegen dispatch | Routed through the JVM codegen dispatcher |
 | `map_concat` | ✅ | Codegen dispatch |  |
 | `map_contains_key` | ✅ | — |  |
 | `map_entries` | ✅ | Native |  |
-| `map_from_arrays` | ✅ | Hybrid | `FLOAT` or `DOUBLE` keys and `spark.sql.mapKeyDedupPolicy=LAST_WIN` route through the JVM codegen dispatcher ([details](compatibility/expressions/map.md)) |
+| `map_from_arrays` | ✅ | Hybrid | A nondeterministic operand, `FLOAT` or `DOUBLE` keys, and `spark.sql.mapKeyDedupPolicy=LAST_WIN` route through the JVM codegen dispatcher ([details](compatibility/expressions/map.md)) |
 | `map_from_entries` | ✅ | Hybrid | BinaryType keys/values, `FLOAT` or `DOUBLE` keys and `spark.sql.mapKeyDedupPolicy=LAST_WIN` route through the JVM codegen dispatcher ([details](compatibility/expressions/map.md)) |
 | `map_keys` | ✅ | Native |  |
 | `map_values` | ✅ | Native |  |

pom.xml

@@ -990,6 +990,7 @@ under the License.
                     <!-- Default extension hooks intentionally leave parameters for overrides. -->
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.rules\.CometScanContrib\.tryTransformV[12]\..*:s</arg>
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CometExpressionSerde\.getSupportLevel\..*:s</arg>
+                    <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CodegenDispatchFallback\.dispatchDeclineReason\..*:s</arg>
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CometAggregateExpressionSerde\.getSupportLevel\..*:s</arg>
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CometOperatorSerde\.getSupportLevel\..*:s</arg>
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CometAggregateExpressionSerde\.supportsSparkPartialToNativeFinal\..*:s</arg>
@@ -1035,6 +1036,7 @@ under the License.
                     <!-- Default extension hooks intentionally leave parameters for overrides. -->
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.rules\.CometScanContrib\.tryTransformV[12]\..*:s</arg>
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CometExpressionSerde\.getSupportLevel\..*:s</arg>
+                    <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CodegenDispatchFallback\.dispatchDeclineReason\..*:s</arg>
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CometAggregateExpressionSerde\.getSupportLevel\..*:s</arg>
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CometOperatorSerde\.getSupportLevel\..*:s</arg>
                     <arg>-Wconf:cat=unused-params&amp;site=org\.apache\.comet\.serde\.CometAggregateExpressionSerde\.supportsSparkPartialToNativeFinal\..*:s</arg>

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -1315,9 +1315,20 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
       // skip a function's arguments (see `QueryPlanSerde.canShortCircuitNulls`). Spark's
       // generated code for an aggregate shares them whatever
       // spark.sql.subexpressionElimination.enabled says, and other operators only when it is on.
+      // A projection shares in generated code. The prefix Spark computes for a sort's first key
+      // is one projection, so it shares within that key as a project does, and later keys are
+      // compared one by one and never share. Spark moves complex grouping keys into a project
+      // below an aggregate, so a key has nothing to skip, and the aggregate's serde opens nested
+      // scopes for its functions' inputs and its results that supersede this one.
+      val (scope, certain) = op match {
+        case s: SortExec => (s.sortOrder.take(1).map(_.child), true)
+        case _: ProjectExec | _: BaseAggregateExec => (op.expressions, true)
+        case _ => (op.expressions, false)
+      }
       val nativeOp = QueryPlanSerde.withSharedSubexpressions(
-        op.expressions,
-        alwaysShared = op.isInstanceOf[BaseAggregateExec]) {
+        scope,
+        alwaysShared = op.isInstanceOf[BaseAggregateExec],
+        certain = certain) {
         if (op.children.nonEmpty && op.children.forall(_.isInstanceOf[CometNativeExec])) {
           val childOp = op.children.map(_.asInstanceOf[CometNativeExec].nativeOp)
           childOp.foreach(builder.addChildren)

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

@@ -20,6 +20,9 @@
 package org.apache.comet.serde
 
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.types.CalendarIntervalType
+
+import org.apache.comet.serde.QueryPlanSerde.skipsSharedSubexpression
 
 /**
  * Trait for providing serialization logic for expressions.
@@ -112,7 +115,8 @@ trait CometExpressionSerde[T <: Expression] {
  * `Unsupported`, the expression always routes through the dispatcher -- the serde is declaring
  * "no native path exists for this case; run Spark's code in-pipeline." Spark fallback is reserved
  * for the case where the dispatcher itself cannot handle the expression (e.g. the global codegen
- * flag is off, or the kernel rejects the bound tree).
+ * flag is off, or the kernel rejects the bound tree), or an expression that
+ * `dispatchDeclineReason` declines.
  *
  * Contract for `Unsupported` reasons on a `CodegenDispatchFallback` serde: the case must be
  * something `Expression.doGenCode` can compile. If you mark something `Unsupported` because Spark
@@ -122,7 +126,43 @@ trait CometExpressionSerde[T <: Expression] {
  * dispatcher. Every other `Incompatible` expression falls back to Spark, and every other
  * `Unsupported` expression falls back to Spark.
  */
-trait CodegenDispatchFallback extends NativeOptInAvailable { self: CometExpressionSerde[_] => }
+trait CodegenDispatchFallback extends NativeOptInAvailable { self: CometExpressionSerde[_] =>
+
+  /** Why `expr` must fall back to Spark rather than run through the dispatcher, or `None`. */
+  def dispatchDeclineReason(expr: Expression): Option[String] = None
+}
+
+/** Sends the two cases below, which the dispatcher would get wrong, to Spark. */
+trait GatedCodegenDispatchFallback extends CodegenDispatchFallback {
+  self: CometExpressionSerde[_] =>
+
+  override def dispatchDeclineReason(expr: Expression): Option[String] =
+    CalendarIntervalOutput
+      .declineReason(expr)
+      .orElse(SharedSubexpressionArgument.declineReason(expr))
+}
+
+/** See https://github.com/apache/datafusion-comet/issues/5279. */
+private[serde] object CalendarIntervalOutput {
+
+  private val reason: String =
+    "the result holds a calendar interval, whose microseconds overflow the dispatcher's Arrow " +
+      "nanoseconds beyond about 292 years"
+
+  def declineReason(expr: Expression): Option[String] =
+    if (SupportLevel.containsType(expr.dataType, classOf[CalendarIntervalType])) Some(reason)
+    else None
+}
+
+private[serde] object SharedSubexpressionArgument {
+
+  private val reason: String =
+    "an argument this expression can skip holds a subexpression that the operator's other " +
+      "expressions share, which Spark evaluates for every row before this expression's NULL check"
+
+  def declineReason(expr: Expression): Option[String] =
+    if (skipsSharedSubexpression(expr)) Some(reason) else None
+}
 
 /**
  * Marker for serdes that have a native implementation the user can opt into. Normally these