fix: defer throwing literal casts to Spark at runtime

[sunchao]

Which issue does this PR close?

No existing issue. This fixes the literal-cast planning failure described below.

Rationale for this change

Spark can retain a throwing ANSI cast of a literal inside a conditional branch that the query never visits. Comet evaluates literal casts while serializing the plan, so it can raise before Spark requests that branch.

For example, with Parquet rows (0, "0") and (1, "1"), this query should return 0:

SELECT CAST(IF(id = 1, 'bad', value) AS INT)
FROM cast_branch_rows
LIMIT 1

Spark pushes the cast into the conditional branches and retains the invalid literal cast for conditional evaluation. Eagerly evaluating it during Comet planning changes the successful query into an error.

What changes are included in this PR?

When evaluation of a literal cast throws a nonfatal exception, return a fallback with an explanation so Spark evaluates the expression when demanded. Successful literal casts keep the existing folding path.

The regression checks the unvisited branch returns the first row while retaining a native Parquet scan. A second query demands the invalid branch and verifies both executions raise CAST_INVALID_INPUT with matching exception type and SQLSTATE.

How are these changes tested?

• Built the native library from current upstream sources with make core.
• Full Spark 4.1 test compilation, Spotless, and Scalastyle pass.
• The new regression fails on the unchanged upstream implementation with CAST_INVALID_INPUT during planning.
• With the fix, the complete CometScalarFunctionSuite passes: 15 tests, no failures or aborted suites (mvn test -Pspark-4.1 -Dtest=none -Dsuites=org.apache.comet.serde.CometScalarFunctionSuite).
• The run-spark-4.1-tests label requests the broader Spark SQL checks in CI.

Codex assisted with adapting the existing fix and regression, source review, and validation.

[viirya]

Thanks for tracking this down — the root cause analysis is accurate and I verified it against Spark's source. ConstantFolding.tryFold deliberately leaves a throwing expression in place when it sits in a conditional branch, tagging it FAILED_TO_EVALUATE rather than folding it:

// sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala
case NonFatal(_) if isConditionalBranch =>
  expr.setTagValue(FAILED_TO_EVALUATE, ())
  expr

So Cast(Literal("bad"), IntegerType) survives into the physical plan, and Comet's unconditional cast.eval() during serialization raises for a branch that may never be visited at runtime. That turns a query Spark answers successfully into a planning-time failure. Real bug, worth fixing, and catching NonFatal matches the range Spark's own tryFold catches, so the two stay aligned on what counts as a recoverable failure.

My main concern is where the fallback happens rather than whether it should.

The fallback bypasses the codegen dispatcher. CometCast mixes in CodegenDispatchFallback, whose contract states that Unsupported means "no native path exists for this case; run Spark's doGenCode inside the Comet pipeline," and that Spark fallback is reserved for cases the dispatcher itself cannot handle. But exprToProtoInternal only calls dispatchIfFallback from the Unsupported and Incompatible branches — Compatible goes straight to convert, so a None returned there falls the whole projection back to Spark without the dispatcher ever being tried.

A throwing literal cast is squarely within what the dispatcher handles: Cast.doGenCode compiles, and it raises at runtime only if the branch is actually visited, which is exactly Spark's behavior. Evaluating in getSupportLevel and returning Unsupported(Some(reason)) on failure would let the framework try the dispatcher first and fall back to Spark only if that fails — strictly better than the current outcome, and it is what this same file already does for VariantType:

// Reporting `Unsupported` lets the `CodegenDispatchFallback` mixin try the dispatcher
// and then fall back to Spark cleanly.

It also removes the inconsistency of getSupportLevel reporting Compatible for an expression convert cannot actually produce. The cost is evaluating the literal twice on the success path, which seems negligible — the failure path never reaches convert.

Worth noting for the record: ConstantFolding.FAILED_TO_EVALUATE would be a more precise signal than catching from eval(), but it is private[sql] and unreachable from org.apache.comet.expressions, so try/catch is a reasonable substitute. A comment saying so would save the next reader from re-deriving it.

I checked for sibling instances of this bug class and found none that need fixing here. I swept every eager .eval( call in the serde layer on main. Round's r.scale.eval(EmptyRow) looked like the closest match, but RoundBase.dataType and checkInputDataTypes already evaluate _scale during analysis, so Spark raises first; the rest (ArraySort's ascendingOrder, the percentile and bloom-filter parameters, window frame bounds) are all analysis-validated foldable arguments. The scope of this PR looks right.

Smaller points, all in inline comments: the fallback reason should be a shared constant to match the convention this file documents, the test duplicates the existing checkSparkError helper in a slightly weaker form, and it is in a suite about something else. A negative test asserting that successful literal casts still fold natively would guard against the catch being widened later.

[viirya]

This is the block that decides the expression falls all the way back to Spark. Because getSupportLevel returned Compatible, exprToProtoInternal took the branch that calls convert directly, so dispatchIfFallback is never reached and the whole projection leaves the Comet pipeline.

Consider moving the decision into getSupportLevel: try the evaluation there, and on NonFatal return Unsupported(Some(literalCastConditionalEvalReason)). The framework then tries the JVM codegen dispatcher first — which can run Cast.doGenCode in-pipeline and raise only if the branch is actually visited — and falls back to Spark only if the dispatcher declines. convert can then keep its current unconditional cast.eval(), since the failing case no longer reaches it.

Two smaller things on this same block:

The reason string should be a shared constant. This file already establishes the convention, with the rationale spelled out at the top:

// Shared with CometNativeCastSuite so the asserted reason cannot drift from production.
private[comet] val negativeScaleDecimalToStringReason: String = ...

Here the string is written literally in both production and the test. Since checkSparkAnswerAndFallbackReasons matches with contains, a future edit would not break compilation and would not necessarily fail the test either — exactly the drift that comment guards against. On the wording: "Literal cast requires Spark's conditional evaluation" states the remedy but not the cause, and a user seeing this in EXPLAIN is asking why their query left Comet. Something self-explanatory would read better, e.g. "Cast of a literal threw during planning; Spark leaves it for conditional evaluation so it may never be reached at runtime".

Please add a comment explaining why the catch exists. The reason is genuinely non-obvious: Spark's ConstantFolding.tryFold catches NonFatal for expressions inside a conditional branch, tags them FAILED_TO_EVALUATE and leaves them unfolded, so a cast that throws can legitimately survive into the plan without ever being evaluated at runtime. Without that context, catch { case NonFatal(_) => return None } reads like swallowing an error. Worth also noting that FAILED_TO_EVALUATE would be the precise signal but is private[sql] and unusable from this package — otherwise someone will reasonably wonder why it is not used.

[comphead]

Agree with @viirya would be nice to have more reasonable fall back message

[viirya]

This suite is entirely CometScalarFunction ANSI-sensitivity unit tests — all 15 call .convert(...) directly, with no SparkSession, no Parquet and no query execution. This test is end-to-end and exercises CometCast, so CometNativeCastSuite looks like the natural home; it is already the canonical place for cast behavior and already imports CometCast constants for exactly this kind of assertion.

[viirya]

The test's validity rests on ConstantFolding refusing to fold CAST('bad' AS INT) because it sits in a conditional branch — but nothing here says so. If Spark's folding behavior ever changes, this test would keep passing while silently no longer covering the bug.

A brief comment, or an assertion that the Cast is still present in the optimized plan, would pin the premise down.

[viirya]

CometTestBase.checkSparkError(df, errorClass) already does all of this, and is stricter in two ways worth keeping:

• it asserts no CometNativeException appears in the cause chain, so the test cannot pass on an error that surfaced from native code with a coincidentally matching class;
• it uses lastOption.getOrElse(fail(...)), whereas .last here throws UnsupportedOperationException if the chain contains no SparkThrowable, turning a meaningful assertion failure into a confusing one.

The whole block collapses to:

checkSparkError(sql(s"SELECT $cast FROM cast_branch_rows"), "CAST_INVALID_INPUT")

which also reads better than the errors(0) / errors(1) indexing.

(Unrelated to this PR: getErrorClass is deprecated in favor of getCondition(), but the shared helper still uses it, so that is better changed there on its own.)

[viirya]

Could you add a case asserting the fix is not over-corrected — that a successful literal cast (say CAST('1' AS INT) in the same branch position) still folds and stays native, with no fallback reason? #5623 paired its guard with exactly this kind of counterpart test, and without one there is nothing stopping the catch from being widened later.

A non-ANSI case would also be worth having, or a note explaining why ANSI coverage is sufficient.