fix: preserve skipped unbase64 rows

.github/workflows/pr_build_linux.yml

@@ -612,6 +612,7 @@ jobs:
               org.apache.comet.CometStringDecodeSuite
               org.apache.comet.CometWidthBucketSuite
               org.apache.comet.CometUuidExpressionSuite
+              org.apache.comet.serde.CometEvaluationMaskSuite
               org.apache.comet.serde.CometScalarFunctionSuite
               org.apache.comet.serde.CometLiteralSuite
               org.apache.comet.CometFallbackInvarianceSuite

.github/workflows/pr_build_macos.yml

@@ -260,6 +260,7 @@ jobs:
               org.apache.comet.CometStringDecodeSuite
               org.apache.comet.CometWidthBucketSuite
               org.apache.comet.CometUuidExpressionSuite
+              org.apache.comet.serde.CometEvaluationMaskSuite
               org.apache.comet.serde.CometScalarFunctionSuite
               org.apache.comet.serde.CometLiteralSuite
               org.apache.comet.CometFallbackInvarianceSuite

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

@@ -253,6 +253,10 @@
 - Spark 4.0.1 (audited 2026-08-24): `NullIntolerant` becomes `override def nullIntolerant: Boolean = true`; `inputTypes` widens to `StringTypeWithCollation(supportsTrimCollation = true)`. Behaviour is byte-level and collation-independent, so no divergence for `UTF8_BINARY` and no shim is needed.
 - Spark 4.1.1 (audited 2026-08-24): adds `contextIndependentFoldable`; no behavioural change on the decode path.
 - Comet native implementation (`spark_unbase64`, `native/spark-expr/src/string_funcs/unbase64.rs`) ports the JDK MIME decoder rules: 256-entry decode LUT, a reused per-batch scratch `Vec<u8>` copied into a preallocated `BinaryBuilder`, all four error messages reproduced verbatim. `CometUnBase64` handles `failOnError = false` natively; `failOnError = true` (reachable from `to_binary('base64')` / `try_to_binary`) requires strict RFC 4648 validation and is not yet implemented natively, so those cases stay on the JVM codegen dispatcher via `CodegenDispatchFallback`.
+- Native decoding is limited to column and literal children. Compound children use JVM codegen dispatch when enabled, otherwise they fall back to Spark. Both native decoding and JVM dispatch evaluate batches, so `CometExecRule` also preserves Spark's row pipeline where a limit, preordered top-K or unpartitioned `WindowGroupLimit`, or a first-match semi/anti join condition can skip malformed input. Partitioned window limits drain their groups and do not independently require this fallback, but an outer limit can still protect them. Ordinary decoding projections below a blocking sort or exchange remain eligible unless another limiting consumer intervenes.
+- With AQE and an outer limit, a final aggregate that decodes its results and uses incompatible intermediate buffers stays in Spark together with its partial aggregate before stages materialize. This also applies when an intervening sort currently consumes all input, because AQE can remove that sort later. A final aggregate only merging already-decoded inputs does not need this result-evaluation protection.
+- `spark.comet.exec.preserveEvaluationMasks.enabled` defaults to `true`. Users who know their Base64 input is valid can set it to `false` to retain native execution across these operator boundaries, including the AQE aggregate protection above. With this opt-out, eager batch evaluation can raise errors on malformed values in rows Spark would skip. Expression support restrictions, including strict decoding and compound-child dispatch, still apply.
+- Remaining evaluation-mask gaps, including other throwing expressions and enclosing filter/conditional masks, are tracked in [#6006](https://github.com/apache/datafusion-comet/issues/6006).
 
 ## upper
 

docs/source/user-guide/latest/compatibility/index.md

@@ -124,6 +124,20 @@ divergence:
 - Spark 4.2 introduced additional ANSI arithmetic overflow behavior differences that Comet does
   not yet track ([#4967](https://github.com/apache/datafusion-comet/issues/4967)).
 
+## Errors from rows Spark skips
+
+Batch evaluation can raise a data-dependent error on a row that Spark's row pipeline would
+never evaluate. Comet preserves operator-level evaluation masks for `unbase64` under limits,
+first-match semi/anti join conditions, and ordered top-K or unpartitioned window limits.
+This protection is enabled by default. Setting
+`spark.comet.exec.preserveEvaluationMasks.enabled=false` retains native execution for known-valid
+input, but can raise errors from malformed Base64 in rows Spark would skip.
+
+This policy currently enrolls only `UnBase64`; it does not establish error-evaluation parity for
+ANSI arithmetic, casts, decimal division, or `element_at`. Per-row conditional evaluation
+(`AND`, `CASE`) and aggregate `FILTER` require separate checks. The broader audit and remaining
+work are tracked in [#6006](https://github.com/apache/datafusion-comet/issues/6006).
+
 ## Known result-value divergences
 
 The following native paths silently return values that differ from Spark for edge-case inputs.

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

@@ -615,7 +615,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | `try_to_binary` | ✅ | — | Runs natively (rewrites to `try_eval(to_binary(...))`) |
 | `try_to_number` | ✅ | Codegen dispatch | Routed through the JVM codegen dispatcher |
 | `ucase` | ✅ | Hybrid |  |
-| `unbase64` | ✅ | Codegen dispatch |  |
+| `unbase64` | ✅ | Codegen dispatch | Native decoding accepts column and literal inputs; strict validation and compound inputs use JVM dispatch when enabled. Where limits, preordered top-K or unpartitioned window limits, or first-match semi/anti join conditions can skip malformed input, the affected pipeline stays in Spark's row execution. For known-valid input, set `spark.comet.exec.preserveEvaluationMasks.enabled=false` to retain native execution. Disabling this protection can raise errors on malformed values in rows Spark would skip. |
 | `upper` | ✅ | Hybrid |  |
 
 ---

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -215,6 +215,17 @@ object CometConf extends ShimCometConf {
     .booleanConf
     .createWithDefault(true)
 
+  val COMET_EXEC_PRESERVE_EVALUATION_MASKS_ENABLED: ConfigEntry[Boolean] =
+    conf(s"$COMET_EXEC_CONFIG_PREFIX.preserveEvaluationMasks.enabled")
+      .category(CATEGORY_EXEC)
+      .doc(
+        "Preserve Spark's row evaluation when operators such as LIMIT or semi/anti joins " +
+          "can skip inputs to throwing expressions. Currently this applies to unbase64. " +
+          "Set to false to retain native execution when the input is known to be valid. " +
+          "When disabled, Comet may fail on malformed Base64 in rows Spark would skip.")
+      .booleanConf
+      .createWithDefault(true)
+
   val COMET_EXEC_PROJECT_ENABLED: ConfigEntry[Boolean] =
     createExecEnabledConfig("project", defaultValue = true)
   val COMET_EXEC_FILTER_ENABLED: ConfigEntry[Boolean] =

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

@@ -22,9 +22,10 @@ package org.apache.comet.rules
 import scala.collection.mutable.ListBuffer
 
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.expressions.{Divide, DoubleLiteral, EqualNullSafe, EqualTo, Expression, FloatLiteral, GreaterThan, GreaterThanOrEqual, KnownFloatingPointNormalized, LessThan, LessThanOrEqual, NamedExpression, Remainder}
+import org.apache.spark.sql.catalyst.expressions.{Divide, DoubleLiteral, EqualNullSafe, EqualTo, Expression, FloatLiteral, GreaterThan, GreaterThanOrEqual, KnownFloatingPointNormalized, LessThan, LessThanOrEqual, NamedExpression, Remainder, SortOrder}
 import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateMode, Final, Partial, PartialMerge}
 import org.apache.spark.sql.catalyst.optimizer.NormalizeNaNAndZero
+import org.apache.spark.sql.catalyst.plans.{JoinType, LeftAnti, LeftSemi}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 import org.apache.spark.sql.catalyst.util.sideBySide
@@ -47,7 +48,7 @@ import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan
 import org.apache.spark.sql.execution.datasources.v2.json.JsonScan
 import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
 import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, BroadcastExchangeLike, ReusedExchangeExec, ShuffleExchangeExec, ShuffleExchangeLike}
-import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, HashJoin, ShuffledHashJoinExec, SortMergeJoinExec}
 import org.apache.spark.sql.execution.window.WindowExec
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
@@ -70,6 +71,9 @@ object CometExecRule {
   val COMET_UNSAFE_PARTIAL: TreeNodeTag[String] =
     TreeNodeTag[String]("comet.unsafePartialAgg")
 
+  private[rules] val UNSAFE_EXPRESSION_EVALUATION: TreeNodeTag[String] =
+    TreeNodeTag[String]("comet.unsafeExpressionEvaluation")
+
   /**
    * Fully native operators.
    */
@@ -339,6 +343,9 @@ case class CometExecRule(session: SparkSession)
   // spotless:on
   private def transform(plan: SparkPlan): SparkPlan = {
     def convertNode(op: SparkPlan): SparkPlan = op match {
+      case op if op.getTagValue(CometExecRule.UNSAFE_EXPRESSION_EVALUATION).isDefined =>
+        withFallbackReason(op, op.getTagValue(CometExecRule.UNSAFE_EXPRESSION_EVALUATION).get)
+
       // Scan marker produced by an optional, out-of-tree scan contrib (e.g. contrib/delta).
       // Matched by trait (no compile-time dependency on the contrib) and present only when that
       // contrib is on the classpath. The marker carries its own serde handler and typically wraps
@@ -764,9 +771,10 @@ case class CometExecRule(session: SparkSession)
       // corresponding Final or PartialMerge cannot be converted and the intermediate buffer
       // formats are incompatible. This runs before transform() so the tags are checked
       // during the bottom-up conversion. Tags persist through AQE stage creation.
-      tagUnsafePartialAggregates(planWithJoinRewritten)
+      val planWithEvaluationMasks = preserveEvaluationMasks(planWithJoinRewritten)
+      tagUnsafePartialAggregates(planWithEvaluationMasks)
 
-      var newPlan = revertUnsafePartialAggregates(transform(planWithJoinRewritten))
+      var newPlan = revertUnsafePartialAggregates(transform(planWithEvaluationMasks))
 
       // if the plan cannot be run fully natively then explain why (when appropriate
       // config is enabled)
@@ -867,6 +875,150 @@ case class CometExecRule(session: SparkSession)
     }
   }
 
+  /** Keep opted-in expressions in Spark's row pipeline where an operator can skip inputs. */
+  private def preserveEvaluationMasks(plan: SparkPlan): SparkPlan = {
+    if (!CometConf.COMET_EXEC_PRESERVE_EVALUATION_MASKS_ENABLED.get()) return plan
+
+    def findEvaluationMaskName(expr: Expression): Option[String] = {
+      var name: Option[String] = None
+      expr.exists { child =>
+        name = QueryPlanSerde.evaluationMaskName(child)
+        name.isDefined
+      }
+      name
+    }
+
+    def originalPlan(node: SparkPlan): SparkPlan = node match {
+      case scan: CometScanExec =>
+        scan.wrapped
+          .copy(partitionFilters = scan.partitionFilters, dataFilters = scan.dataFilters)
+      case comet: CometExec => comet.originalPlan
+      case shuffle: CometShuffleExchangeExec => shuffle.originalPlan
+      case broadcast: CometBroadcastExchangeExec => broadcast.originalPlan
+      case _ => node
+    }
+
+    // Most plans contain no opted-in expression. Include native originals and sticky tags
+    // so repeated AQE passes still restore an already-protected subtree when necessary.
+    if (!plan.exists(node =>
+        node.getTagValue(CometExecRule.UNSAFE_EXPRESSION_EVALUATION).isDefined ||
+          originalPlan(node).expressions.exists(findEvaluationMaskName(_).isDefined))) {
+      return plan
+    }
+
+    def firstMatch(joinType: JoinType): Boolean = joinType match {
+      case LeftSemi | LeftAnti => true
+      case _ => false
+    }
+
+    def protect(
+        node: SparkPlan,
+        belowLimit: Boolean,
+        hasLimitAncestor: Boolean): (SparkPlan, Option[String]) = {
+      val original = originalPlan(node)
+      val startsLimit = original match {
+        // Offset-only collection does not stop its input early.
+        case collect: CollectLimitExec => collect.limit >= 0
+        case _: LocalLimitExec | _: GlobalLimitExec => true
+        case topK: TakeOrderedAndProjectExec =>
+          SortOrder.orderingSatisfies(node.children.head.outputOrdering, topK.sortOrder)
+        case windowLimit
+            if ShimCometWindowGroupLimit.windowGroupLimitClass.exists(
+              _.isInstance(windowLimit)) =>
+          // Partitioned limits drain each group, but an outer LIMIT can still stop them.
+          // Keep the conservative behavior if a future rank function cannot be extracted.
+          ShimCometWindowGroupLimit.extract(windowLimit).forall(_.partitionSpec.isEmpty) &&
+          SortOrder.orderingSatisfies(
+            node.children.head.outputOrdering,
+            windowLimit.requiredChildOrdering.head)
+        case _ => false
+      }
+      // These operators consume their input before yielding rows. Still visit their children:
+      // an inner LocalLimit below an exchange must establish its own evaluation boundary.
+      val materializesInput = original match {
+        case _: SortExec | _: HashAggregateExec | _: ObjectHashAggregateExec |
+            _: ShuffleExchangeLike | _: BroadcastExchangeLike | _: QueryStageExec |
+            _: ReusedExchangeExec =>
+          true
+        case _ => false
+      }
+      // A top-K is still a LIMIT even when its current input requires sorting.
+      val limitAncestor = hasLimitAncestor || startsLimit ||
+        original.isInstanceOf[TakeOrderedAndProjectExec]
+      val protectedChildren = node.children.map { child =>
+        protect(child, startsLimit || (belowLimit && !materializesInput), limitAncestor)
+      }
+      val childReason = protectedChildren.flatMap(_._2).headOption
+      val condition = original match {
+        case join: HashJoin if firstMatch(join.joinType) => join.condition
+        case join: SortMergeJoinExec if firstMatch(join.joinType) => join.condition
+        case join: BroadcastNestedLoopJoinExec if firstMatch(join.joinType) => join.condition
+        case _ => None
+      }
+      val finalAggregate = original match {
+        case agg: BaseAggregateExec
+            if (agg.isInstanceOf[HashAggregateExec] ||
+              agg.isInstanceOf[ObjectHashAggregateExec]) &&
+              agg.aggregateExpressions.map(_.mode).distinct == Seq(Final) =>
+          Some(agg)
+        case _ => None
+      }
+      val limitName = if (belowLimit) {
+        // Final merges buffers; it does not reevaluate the aggregate's original inputs.
+        val expressions = finalAggregate.map(_.resultExpressions).getOrElse(original.expressions)
+        expressions.iterator.flatMap(findEvaluationMaskName).take(1).toSeq.headOption
+      } else {
+        None
+      }
+      // AQE can remove an intervening sort after a native Partial has materialized. Choose
+      // compatible buffers before that happens, even if a current operator drains its input.
+      val aggregateBufferName = if (hasLimitAncestor && conf.adaptiveExecutionEnabled) {
+        finalAggregate
+          .filterNot(agg =>
+            QueryPlanSerde.allAggsSupportNativePartialToSparkFinal(agg.aggregateExpressions))
+          .flatMap(
+            _.resultExpressions.iterator.flatMap(findEvaluationMaskName).take(1).toSeq.headOption)
+      } else {
+        None
+      }
+      val ownReason = limitName
+        .map(name => s"$name requires Spark evaluation below LIMIT")
+        .orElse(aggregateBufferName.map(name =>
+          s"$name requires Spark aggregate buffers below LIMIT with AQE"))
+        .orElse(
+          condition
+            .flatMap(findEvaluationMaskName)
+            .map(name => s"$name requires Spark evaluation in first-match join conditions"))
+        .orElse(node.getTagValue(CometExecRule.UNSAFE_EXPRESSION_EVALUATION))
+      // Exchanges can restart native execution after consuming a Spark row pipeline.
+      val restartsNative = original.isInstanceOf[ShuffleExchangeLike] ||
+        original.isInstanceOf[BroadcastExchangeLike]
+      val reason = ownReason.orElse(if (restartsNative) None else childReason)
+      // Early tags keep fresh aggregate buffers safe before AQE materialization. Reused native
+      // buffer chains are restored by revertUnsafePartialAggregates after conversion.
+      val children = protectedChildren.map(_._1)
+      val prepared = node match {
+        // Do not refill a batch between the row decoder and its short-circuiting consumer.
+        case _: RowToColumnarExec | _: CometSparkToColumnarExec if childReason.isDefined =>
+          children.head
+        case _: ColumnarToRowExec | _: CometColumnarToRowExec | _: CometNativeColumnarToRowExec
+            if childReason.isDefined && !children.head.supportsColumnar =>
+          children.head
+        case _ if (original ne node) && (reason.isDefined || children != node.children) =>
+          // AQE can reuse an existing native subtree. Rebuild affected ancestors as well so
+          // their serialized native plans do not retain the decoder that just fell back.
+          val restored = original.withNewChildren(children)
+          node.getTagValue(SparkPlan.LOGICAL_PLAN_TAG).foreach(restored.setLogicalLink)
+          restored
+        case _ => node.withNewChildren(children)
+      }
+      reason.foreach(prepared.setTagValue(CometExecRule.UNSAFE_EXPRESSION_EVALUATION, _))
+      (prepared, reason)
+    }
+
+    protect(plan, belowLimit = false, hasLimitAncestor = false)._1
+  }
+
   /** Convert a Spark plan to a Comet plan using the specified serde handler */
   private def convertToComet(op: SparkPlan, handler: CometOperatorSerde[_]): Option[SparkPlan] = {
     val converted = tryConvertToComet(op, handler)
@@ -1318,6 +1470,8 @@ case class CometExecRule(session: SparkSession)
   private def canAggregateBeConverted(
       agg: BaseAggregateExec,
       expectedMode: AggregateMode): Boolean = {
+    if (agg.getTagValue(CometExecRule.UNSAFE_EXPRESSION_EVALUATION).isDefined) return false
+
     val handler = allExecs.get(agg.getClass)
     if (handler.isEmpty) return false
     val serde = handler.get.asInstanceOf[CometOperatorSerde[SparkPlan]]

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

@@ -124,6 +124,20 @@ trait CometExpressionSerde[T <: Expression] {
  */
 trait CodegenDispatchFallback extends NativeOptInAvailable { self: CometExpressionSerde[_] => }
 
+/**
+ * Opts an expression instance into the planner's existing operator-level evaluation protection.
+ * This is not an exhaustive classification of expressions that can throw, nor a guarantee that
+ * all evaluation contexts are protected. The policy is independent of native or JVM-dispatch
+ * support and must inspect the expression without evaluating it or serializing it.
+ *
+ * Return a stable function name for fallback explanations, or None when the instance is not
+ * enrolled. None does not prove safety. Future enrollment requires separate compatibility and
+ * performance tests, including evaluation mode, input types, and enclosing expressions.
+ */
+trait RequiresSparkEvaluationMask[T <: Expression] { self: CometExpressionSerde[T] =>
+  def evaluationMaskName(expr: T): Option[String]
+}
+
 /**
  * Marker for serdes that have a native implementation the user can opt into. Normally these
  * serdes run a Spark-compatible path by default. A serde can set `hasConditionalNativeDefault`