fix: preserve Spark evaluation for next_day and levenshtein

[sunchao]

Which issue does this PR close?

Follow-up to #5591 and #5720.

Rationale for this change

Running an expression through Spark's generated evaluator does not always preserve how the enclosing Spark operator evaluates it. In particular, interpreted three-argument levenshtein treats a NULL threshold as zero, while generated evaluation returns NULL. For rows ('', '', NULL), ('a', 'b', NULL), and ('a', 'b', 1), collect_list(levenshtein(...)) must retain [0, -1, 1]; generating its arguments separately drops the first two values. The same distinction affects NO_CODEGEN, array_compact, and approx_count_distinct.

ANSI next_day can also raise errors for rows or operands Spark skips. A nullable comparison, a LIMIT, or a first-match join can avoid an invalid weekday. Whole-stage execution can defer a Project or grouped-aggregate result until a filter or conditional consumes it, and try_add can catch an error raised inside that deferred expression. Materializing it in a Comet batch changes those results. Separately, a foldable weekday can fail during code generation where Spark's projection would recover by interpreting a NULL date.

What changes are included in this PR?

• Preserve interpreted Levenshtein arguments in the affected execution modes and aggregate/array paths. Safe nonnullable thresholds and generated aggregate arguments remain eligible for acceleration.
• Check the complete expression tree before an enclosing dispatcher can hide a NextDay generation failure or a nullable-threshold Levenshtein.
• Preserve ANSI NextDay's nullable-parent, aggregate-argument, LIMIT, first-match join, and deferred-result boundaries. Retain native execution for eager inputs, blocking input stages, non-ANSI expressions, and offset-only collection.
• Adapt the LIMIT/window/AQE buffer handling from fix: preserve skipped unbase64 rows #5533 for NextDay. Protect incompatible aggregate buffers before AQE can remove a sort, and preserve Spark joins when forced native join rewriting would otherwise replace them.
• Add focused SQL regressions and planner assertions, including consumed-error controls, native-admission controls, AQE reuse, and TryEval. Existing feature benchmarks and ordinary collation tests remain in place.

How are these changes tested?

• Spark 4.0.4 root reactor: 51 tests passed, 0 failed; BUILD SUCCESS. Runs the full CometExecRuleSuite and all next_day and levenshtein SQL fixtures. Production/test compilation, Spotless, Scalastyle, and git diff --check passed.
• Eight paired runtime controls reproduced the mismatches using the base implementations from 4c2ab9686 and matched Spark with the patched implementations. The controls cover nullable comparison, deferred Filter, both NextDay projection modes, Levenshtein NO_CODEGEN, array compact, and both imperative aggregates. Class origins were checked, and both runs used the same native library.
• Native-backed tests used the official Linux artifact from Apache CI run 35014298902, commit 347d8cf3d. Its native tree and build inputs exactly match base 4c2ab9686; the intervening commit changes documentation only. The artifact digest was verified. A local native rebuild was blocked because the configured registry lacks the pinned DataFusion 55.1.0 release.
• Independent review of the serializer and operator adaptations found no remaining actionable issues. Other Spark profiles were not run locally.

[rich7420]

Please exclude provably safe instances from this guard. On Spark 4.1 with ANSI enabled, SELECT next_day(d, 'Monday') FROM t LIMIT 1 changes from CometProject and CometCollectLimit with the base rule to Spark Project and CollectLimit on this head. Here d is a date column and Monday is a valid literal, so the invalid-weekday error cannot occur. Please keep this case native and add an admission control alongside the skipped-invalid-weekday regression.

[rich7420]

@sunchao thanks for the patch!

[andygrove]

Thanks for digging into this. I confirmed the levenshtein divergence against the Spark 4.0.1 source and it is real: eval unboxes a Some(null) threshold to 0 while doGenCode wraps the body in ctx.nullSafeExec and leaves the result NULL. Our native kernel returns NULL, so it matches the generated path and diverges wherever Spark interprets. There is definitely something to fix here. My concerns are about the shape of the fix, plus one regression I measured that I would like resolved before this goes in.

This duplicates #5533

preserveNextDayEvaluationMasks is close to line-for-line identical to preserveEvaluationMasks in #5533. The firstMatch, startsLimit, materializesInput, limitAncestor, finalAggregate, limitName, aggregateBufferName, restartsNative and prepared blocks are the same apart from whitespace. The one real substitution is that #5533 looks the expression up through RequiresSparkEvaluationMask and this hardcodes case nextDay: NextDay if nextDay.failOnError.

Both PRs are open, both are yours, and both edit the same lines, so whichever lands second is going to be an unpleasant reconciliation. Could this rebase onto #5533 and enroll NextDay in the policy instead of growing a second traversal? That picks up the config and the compatibility-guide section for free. It would also be good to reference #5533 and #6006 in the description, since #6006 reads like the tracking issue for exactly this family.

Three things this copy dropped that I think are worth keeping:

The early exit. #5533 returns the plan unchanged when no node carries an enrolled expression or a sticky tag. Without it protect walks every node of every plan and at each one computes original.expressions, a collectFirst per expression tree, supportsWholeStage (another full expression walk plus two isTooManyFields calls) and eagerReferences. Every query pays that, next_day or not.

The config. #5533 added spark.comet.exec.preserveEvaluationMasks.enabled. There is no way to turn this one off, which matters given the next section.

The shared restoreNativeAggregateBuffers. #5533 factors one version shared with the Celeborn fallback. This adds a third local copy, which also drops the withFallbackReason tagging the shared one does.

The fallback blast radius is wider than it needs to be

I built the branch and main and ran the same probe on each, Spark 4.1 default profile, spark.sql.ansi.enabled=true, parquet table, AQE off:

query	main	this PR
next_day(d, dow)	CometProject	CometProject
date_add(next_day(d, dow), 1)	CometProject	Spark Project
datediff(next_day(d, dow), d)	CometProject	Spark Project
concat(cast(next_day(d, dow) AS string), 'x')	CometProject	Spark Project
greatest(next_day(d, dow), d)	CometProject	Spark Project
ORDER BY next_day(d, dow)	CometSort + CometExchange	Spark Sort + Spark Exchange
WHERE k = 1 AND next_day(d, dow) > date'2020-01-01'	CometFilter	Spark Filter
if(k = 1, next_day(d, dow), null)	CometProject	CometProject
GROUP BY next_day(d, dow)	2x CometHashAggregate	2x CometHashAggregate
max(next_day(d, dow))	2x CometHashAggregate	Spark partial + Comet final

Only the AND row is a genuine mask. Spark's generated code for DateAdd, DateDiff, Concat and Greatest evaluates the next_day operand on every row, and a sort key is computed for every row.

The cause is the default arm of unsupportedNextDayEvaluation, case _ => pushChildren(Some(current.nodeName)). Any expression with two or more children outside the five-case whitelist marks all of its children unsafe, including the first child that a nullIntolerant parent always evaluates. eagerReferences in this same PR already gets that right with take(if (expr.children.head.nullable) 1 else 2), so the two halves disagree about date_add. Could they share the rule? And for the ORDER BY case, could SortOrder push None for its child and skip sameOrderExpressions, which are never serialized anyway?

ANSI defaults to on in Spark 4, so this is the default configuration, and none of these shapes has a test or a mention in the description.

The aggregate guard is unconditioned

The check in aggExprToProto declines every Partial and Complete aggregate whose arguments contain an ANSI next_day. The reason text names FILTER, null-skipping and state-dependent updates, but the code checks none of them, and aggExpr.filter is never read. That is what costs max(next_day(...)) its partial aggregate above, and Max's update expression is greatest(max, child), which has no branch. The Levenshtein check directly below it is properly scoped to ImperativeAggregate. Could the NextDay arm be narrowed the same way, to aggExpr.filter.isDefined plus the aggregates that actually skip their child such as First and Last?

levenshtein: worth reporting upstream, and the FALLBACK path is still open

Since eval and doGenCode genuinely disagree, this is a Spark bug rather than a Comet one. Could you file it and link the JIRA here so the workaround has an expiry?

The guard covers CODEGEN_FACTORY_MODE=NO_CODEGEN, which is internal and test-only. What about the default FALLBACK, where generated code that fails to compile drops silently to InterpretedUnsafeProjection? Spark returns 0 there and we return NULL, and nothing catches it. Given how narrow a nullable three-argument levenshtein is, is a plain Unsupported for that shape the better trade than trying to predict which contexts interpret?

Smaller things

wholeExpressionDispatch singles out UnBase64. The reasoning holds for any CodegenDispatchFallback serde whose instance reports Unsupported, which includes CometNextDay and CometLevenshtein themselves. Could that be the general test rather than one expression name hardcoded inside a function about another?

SQLConf.CODEGEN_FACTORY_MODE.toString compared against "NO_CODEGEN" now appears in both CometExecRule and QueryPlanSerde, with the Spark 4.2 explanation comment on only one of them. Can that move into a shim or a single shared helper?

The CometLevenshtein collation reason change reverts the wording #5720 introduced sixteen days ago. That PR's description says the longer text was chosen so it would not contradict the GenerateDocs "no native implementation and always run in the JVM" header. The new string restates the header and drops the raw-bytes explanation. Was that deliberate?

No user-facing documentation. #5533 added an "Errors from rows Spark skips" section to compatibility/index.md for this behavior class. This changes when ANSI next_day is accelerated under Spark 4 defaults with nothing to tell users about it. Extending #5533's section would be better than adding a second one.

What I would suggest

Split it. The levenshtein work is self-contained and could land on its own once the FALLBACK gap is settled. The ANSI next_day work should rebase onto #5533 and go through RequiresSparkEvaluationMask. The deferred-projection analysis is the genuinely new and riskiest piece, it applies to every enrolled expression rather than just next_day, and I think it deserves its own PR against the general policy where it can be justified and tested on its own terms.

On my side: this needs a rebase, and note that main has gained revertUnsafePartialAggregates and the CometRule composition from #6082 inside the code you are editing, so the clean auto-merge is not proof of much. I will add run-spark-sql since this touches both the serde and the planner and the PR tier does not cover it.