fix: preserve Spark evaluation for next_day and levenshtein

.github/workflows/pr_build_linux.yml

@@ -621,6 +621,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.serde.QueryPlanSerdeSuite

.github/workflows/pr_build_macos.yml

@@ -325,6 +325,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.serde.QueryPlanSerdeSuite

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

@@ -119,6 +119,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.
@@ -139,3 +153,19 @@ so users hunting an unexpected value have a single place to check:
   `spark.sql.codegen.fallback` or when the generated code exceeds
   `spark.sql.codegen.hugeMethodLimit`, where an intermediate overflow that later cancels out
   returns `NULL` (or raises under ANSI) in Spark but the recovered value in Comet.
+
+### `next_day` and nullable `levenshtein` thresholds
+
+ANSI `next_day` participates in the evaluation-mask policy above. A valid literal weekday
+stays native, including below a limit. Dynamic weekdays stay in Spark when a limit, first-match
+join, or deferred projection can skip an invalid weekday; eager operands and ordinary sort keys
+remain eligible for native execution. The same `spark.comet.exec.preserveEvaluationMasks.enabled`
+setting controls the operator-level policy. Per-expression guards still preserve conditional
+and filtered/state-dependent aggregate evaluation. The broader family is tracked in
+[#6006](https://github.com/apache/datafusion-comet/issues/6006), and the shared policy is introduced
+by [#5533](https://github.com/apache/datafusion-comet/pull/5533).
+
+Three-argument `levenshtein` with a nullable threshold stays in Spark because Spark's interpreted
+and generated evaluators disagree for NULL thresholds. This applies to every code-generation
+mode, including `FALLBACK`, whose evaluator can change after a compilation failure. Two-argument
+calls and non-nullable thresholds remain eligible for native execution.

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

@@ -625,7 +625,7 @@ to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enab
 | `try_to_binary` | ✅ | — | Rewrites to `try_eval(to_binary(...))`, which routes through the JVM codegen dispatcher |
 | `try_to_number` | ✅ | Codegen dispatch | Routed through the JVM codegen dispatcher |
 | `ucase` | ✅ | Hybrid |  |
-| `unbase64` | ✅ | Codegen dispatch | Column or literal input runs natively; other child expressions, and the strict form used by `to_binary(str, 'base64')`, route through the JVM codegen dispatcher |
+| `unbase64` | ✅ | Codegen dispatch | Column or literal input runs natively; other child expressions, and the strict form used by `to_binary(str, 'base64')`, route through the JVM codegen dispatcher. 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

@@ -260,6 +260,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] =