feat: partial project fallback (JVM expression detour)

[schenksj]

Which issue does this PR close?

Closes #4825.

Rationale for this change

When one expression in a ProjectExec / FilterExec has no native translation, Comet abandons the whole operator and with it the native island below (a ColumnarToRow transition materializes the scan output; supported sibling expressions run row-wise in Spark). This PR evaluates only the unsupported subexpression in the JVM and keeps the operator — and the pipeline — native, by routing that subtree through the codegen dispatcher / JvmScalarUdfExpr callback that already ships on main. No proto or native changes.

This is experimental and default-off (spark.comet.exec.jvmDetour.enabled = false).

What changes are included in this PR?

• CometConf: new spark.comet.exec.jvmDetour.enabled (COMET_EXEC_JVM_DETOUR_ENABLED), default false.
• QueryPlanSerde: a withJvmDetour thread-local gate (a plain non-inheritable ThreadLocal, not DynamicVariable, to avoid leaking the flag to child planning threads) plus a last-resort .orElse hook in exprToProtoInternal that retries any unsupported node via emitJvmCodegenDispatch. On a successful detour it tags the node withInfo and clears its speculative FALLBACK_REASONS so a detoured node isn't also labeled a fallback.
• CometScalaUDF: emitJvmCodegenDispatch now refuses a tree containing a subquery (PlanExpression). A subquery inside the detoured tree is closure-serialized at plan time, before waitForSubqueries populates its result, so evaluating it in the kernel would fail; refusing makes it a clean fallback. (This also hardens the pre-existing ScalaUDF / regex / HOF / date-time dispatch paths, which shared the same latent hazard.)
• operators.scala: wrap CometProjectExec / CometFilterExec convert bodies in withJvmDetour, scoping the detour to those two contexts in v1.

How it works

The hook fires at the outermost unsupported node (serde recursion is handler-driven). Supported ancestors stay native and reference the detour's output; the detoured subtree's argument columns cross to the JVM over Arrow FFI and are evaluated by Spark's own doGenCode / eval inside a Janino kernel, so results match Spark by construction. Plan-time eligibility uses the dispatcher's existing canHandle (rejects aggregates, generators, Unevaluable, unsupported types, oversized trees). If the detour can't fire, None propagates and the operator falls back exactly as before.

How are these changes tested?

• CometPartialProjectFallbackSuite (16 tests): flag-on stays native / flag-off falls back; outermost-node granularity; filter-predicate detour; canHandle-refused clean fallback; serde-level gate scoping; parity through the hook for NULL-heavy input, empty batches, decimals, timestamps, nested array/map, and a nondeterministic tree (per-partition seeding); subquery-inside-tree falls back while a sibling subquery stays native; a mixed projection that falls back whole when one sibling can't detour; island preservation (no columnar→row transition below the native projection).
• CometCodegenFuzzSuite: two added tests fuzz a disabled abs serde through the hook over every numeric column of a random schema and over the decimal precision/scale/null-density sweep.
• CometPartialProjectFallbackBenchmark: Spark vs. detour-off vs. detour-on across three shapes.

CometCodegenSuite (73), CometCodegenFuzzSuite (30), and the new suite (16) all pass locally under the spark-4.1 profile.

Benchmark

16M-row projections/filter over a native Parquet scan, Abs's serde disabled to stand in for an unsupported expression (only the detour flag differs between the two Comet cases). Local Apple M1:

Shape	Spark (best)	Comet, detour off (best)	Comet, detour on (best)	detour-on vs Spark
lone unsupported expr, wide passthrough	414 ms	419 ms	310 ms	1.3×
mixed projection (7 native + 1 unsupported)	434 ms	429 ms	370 ms	1.2×
filter predicate with unsupported subexpr	350 ms	364 ms	278 ms	1.3×

Detour-on preserves the native island (verified by a plan check in the benchmark) and beats both Spark and the whole-operator fallback on all three shapes.

Notes / follow-ups

• Filters: a detoured sub-predicate is evaluated for all rows, the same as any native Comet predicate; under ANSI this can differ from Spark's per-row AND/OR short-circuit. Documented in the config doc; default-off.
• v2 follow-up: rewrite supported children (including subqueries) of the detoured node to native BoundReference args so only their values cross the FFI — this is the path to real subquery support and finer granularity.

🤖 Generated with Claude Code

[andygrove]

To make sure I understand, this PR does not add any new execution mechanism, it just changes the routing.

Prior to this PR, we only support codegen dispatch for a list of expressions that we have deliberately implemented serde and tests for. Essential we have an allow-list today. This PR adds an experimental feature to allow any expression to get routed to codegen dispatch.

@schenksj does that seem correct?

@mbutrovich this sounds similar to a feature you were suggesting recently?

[schenksj]

@andygrove Yes — that's exactly right, with one refinement on "any."

No new execution mechanism. The JVM-detour machinery already ships end-to-end on main — JvmScalarUdfExpr (a DataFusion PhysicalExpr that exports its args over Arrow FFI and calls back into the JVM), CometUdfBridge, and the CometScalaUDFCodegen dispatcher that Janino-compiles Spark's own doGenCode into an Arrow-direct batch kernel. Today that path is reached only for a curated allow-list: ScalaUDF, serdes extending CometCodegenDispatch (HOFs, CSV/JSON, parts of date/time), and Incompatible serdes tagged CodegenDispatchFallback. An expression with no serde handler, or whose convert returns None, aborts the whole operator and drops the native island below it. This PR adds a last-resort .orElse hook so that unsupported node detours to the JVM instead. No proto change, no new JNI surface, no new operator.

On "any expression" — gated, not unconditional. The hook only fires when the existing dispatcher canHandle accepts the subtree, which rejects aggregates, generators, Unevaluable, unsupported types, and oversized trees (and this PR adds a subquery guard). It's also scoped to CometProjectExec / CometFilterExec in v1 — deliberately kept out of join keys, sort keys, group/agg, and partitioning expressions until benchmarked. So: within Project/Filter, any codegen-capable expression can detour, rather than only the allow-listed ones — but it's not a blanket "everything routes to codegen."

Ecosystem precedent (since this is the "does it disrupt planned work" question): this is the same lever Gluten ships as ColumnarPartialProjectExec (on by default, spark.gluten.sql.columnar.partial.project) and Blaze/Auron ship as their partial-project split. The difference is that Comet's version needs no dedicated operator — the detoured subtree is just one more PhysicalExpr in DataFusion's existing ProjectionExec/FilterExec vector, so there's no per-operator batch recomposition or extra C2R/R2C transition. It's default-off (spark.comet.exec.jvmDetour.enabled=false) and experimental, so it can't affect anyone who doesn't opt in.

@mbutrovich — happy to align this with whatever you were sketching; if your idea was operator-shaped (a CometPartialProjectExec), the design doc has a "considered alternatives" section on why v1 went expression-granular instead, and I'd value your read on it.

(Disclosure: this PR and this response were written with help from AI — Claude Code.)