Codegen dispatcher is unreachable from `convert`, so serdes that decline there never get a dispatch attempt

[andygrove]

Part of #5572.

The dispatcher is only reachable from getSupportLevel. dispatchIfFallback is called from exactly two places, the Unsupported arm (spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:941) and the non-opt-in Incompatible arm (:968). The Compatible arm just calls handler.convert(...) and propagates whatever it returns.

So a serde that reports Compatible and then declines inside convert gets no dispatch attempt, even when it mixes in CodegenDispatchFallback. Around 15 serdes decline that way via withFallbackReason, including two that already carry the mixin and therefore silently lose it on that path:

• CometJsonToStructs (serde/structs.scala:195)
• CometArrayReverse (serde/arrays.scala:598)

and, without the mixin, CometCreateArray, CometGetArrayItem, CometGetArrayStructFields, CometArrayInsert, CometStaticInvoke, CometUuid, CometShuffle, CometAttributeReference, CometScalarFunction.

CometUnixTimestamp is the clearest illustration: it makes the identical input-type check twice, once in getSupportLevel (serde/datetime.scala:319) and once in convert (:329). Only the first copy would benefit from adding the mixin.

Two ways to close this:

1. Move the checks into getSupportLevel serde by serde. Precise, no behavior change anywhere else, but it is a lot of small edits and nothing stops the pattern coming back.
2. Add .orElse(dispatchIfFallback(handler, expr, inputs, binding).map(_._2)) after handler.convert in the Compatible arm. One line, and it fixes the pattern permanently.

Option 2 needs care. A Compatible serde's convert can legitimately return None because a child failed to serialize, not because of anything wrong with this node, and in that case retrying the whole subtree through the dispatcher is a real behavior change rather than a bug fix — though arguably a desirable one, since the dispatcher can evaluate a subtree containing an expression that has no native serde at all. It also needs the guard in the prerequisite issue first. Worth prototyping option 2 and measuring how often it changes a plan across the TPC-DS and Spark SQL suites before committing to it.

Related: a warning when getSupportLevel returns Compatible and convert then returns None would make the size of this surface visible, since that combination is a serde invariant violation whenever the cause is the node itself rather than a child.

[andygrove]

Opened #5593 with a warning for the Compatible + convert returns None combination. It does not fix anything here — it just makes the affected cases visible so the cleanup in this issue can be scoped from evidence rather than from a static sweep.

Two things it already turned up that are worth recording against this issue. The suppression rule has to key off descendants only, because CometCreateArray, CometCreateNamedStruct and CometArraysZip tag themselves when a child fails, so a node's own tag says nothing about whose fault the decline was. And with spark.comet.exec.scalaUDF.codegen.enabled=false, every serde that reports Compatible and then routes to emitJvmCodegenDispatch trips the warning — the disabled-config check lives in the dispatcher rather than in getSupportLevel, so that whole family is a real instance of this invariant violation. Moving that check up into CometCodegenDispatch.getSupportLevel looks like the cheapest slice of this issue to do first.

[andygrove]

Prototyped the .orElse variant and measured it. It is unsound as specified — I'd drop it in this form. Details below, since the failure mode is more interesting than the result.

What I built

handler.convert(...).orElse { CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding) } in the Compatible arm, behind an internal off-by-default flag. Deliberately not gated on CodegenDispatchFallback: the mixin marks a serde's declared unsupported cases, whereas reaching here means the serde promised it could convert and then couldn't, which is exactly the case no serde has declared. That ungated form is what would subsume #5575.

What fired

132 rescues across CometExpressionSuite, the array / map / string / temporal / json / math expression suites, CometSqlFileTestSuite and CometCodegenSuite:

count	expression
86	alias
33	staticinvoke
3	element_at
2 each	size, precisetimestampconversion, cast, casewhen, add

The 33 staticinvoke rescues confirm #5575 is subsumed. The 86 alias rescues are the whole-projection case — CometAlias.convert just delegates to its child, so any projection whose tree fails anywhere gets its entire top-level expression dispatched.

The problem

35 test failures. About 33 are tests asserting a fallback that no longer happens — expected, and they'd need updating. Two are wrong answers, both map_contains_key over a nested map with floating-point or collated keys: Spark returns true, Comet returns false.

The mechanism generalises, which is why this matters. For

array_contains(map_keys(element_at(map(1, map(0.0, 7)), k)), -0.0)

CometElementAt.getSupportLevel sees an outer key type of IntegerType and says Compatible. convert then fails anyway, because the nested map literal is unsupported. The rescue dispatches element_at alone, which faithfully produces the inner map — and then map_keys and array_contains run natively on top of it, where array_contains does not normalise -0.0 to +0.0 the way Spark does. Hence false.

So the decline was not protecting the node that declined. It was protecting that node's ancestors, which are only Spark-compatible as long as the whole subtree falls back together. Rescuing at the failing node converts a whole-operator fallback into a local dispatch and silently removes that protection.

That is a general soundness hazard, not a quirk of this one expression: any time a native ancestor's compatibility depends on a descendant having declined, a node-local rescue breaks it. Nothing in the serde API records that dependency today, so there is no way to detect it at the point of rescue.

Where that leaves the design

• Node-local rescue: unsound, drop it.
• Root-level-only rescue (rescue at the Alias / top of the projection, never below): plausible, since the entire expression then runs Spark's own code. But it needs an explicit rule that a rescue may only happen where no native ancestor will consume the result, and the 86 alias hits suggest that is the common case anyway — so most of the upside may survive.
• Either way this does not subsume Route unrecognized StaticInvoke and Invoke through the codegen dispatcher instead of falling back #5575 for free. staticinvoke sits mid-tree as often as not, so the StaticInvoke catch-all is better done as its own targeted change with the type gate doing the safety work.

Practical consequence for the epic: I'd stop treating #5574 as the item that might make the per-expression mixin issues unnecessary. The mixin route reports Unsupported from getSupportLevel, which routes through dispatchIfFallback before any ancestor has converted, so it does not have this hazard. The per-expression issues (#5575–#5591) should proceed on their own.

Prototype is on a local branch; happy to push it if anyone wants to reproduce.