sunchao commented on code in PR #4827:
URL: https://github.com/apache/datafusion-comet/pull/4827#discussion_r4104781363
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -847,6 +886,33 @@ object QueryPlanSerde extends Logging with CometExprShim
with CometTypeShim {
None
}
})
+ .orElse {
+ // Last-resort JVM expression detour. Any `None` above — no registered
handler, an
+ // `Unsupported`/opt-out `Incompatible` support level, or a handler
whose `convert`
+ // returned `None` — means this node has no native translation. Inside
a detour-eligible
+ // context (projection / filter, see `jvmDetourAllowed`) retry the
*outermost* unsupported
+ // node as a JVM detour: `emitJvmCodegenDispatch` ships its argument
columns over Arrow FFI
+ // and evaluates it with Spark's own codegen inside the Comet
pipeline, so the operator —
+ // and the native island around it — stays native. Supported ancestors
keep converting
+ // natively and reference this detour's output. If the detour cannot
fire (feature off,
+ // dispatcher off, or `CometBatchKernelCodegen.canHandle` rejects the
tree), `None`
+ // propagates unchanged and the operator falls back exactly as before.
+ if (jvmDetourAllowed) {
+ CometScalaUDF.emitJvmCodegenDispatch(expr, inputs, binding).map {
proto =>
Review Comment:
[P2] Preserve independent state for newly rescued nondeterministic
expressions. On Spark 4.x, enable the detour and codegen flags and run
`parquetDf.select(expr("uuid(42)").as("a"), expr("uuid(42)").as("b"))` over
three rows in one partition. Spark produces equal values in both columns. These
previously unsupported `Uuid` nodes now serialize identically and share one
mutable kernel through the task-scoped dispatcher cache, so column `b`
continues column `a`'s random stream instead. This silently changes seeded
results. Include expression-occurrence identity in the live-kernel cache key,
or keep nondeterministic rescue disabled until state is isolated.
Evidence: The disposable exact-head harness at
`/tmp/comet-4827-probe/src/ReviewProbe.scala` confirmed identical serialized
bytes for the two parsed, fallback-tagged expressions and `canHandle=None`. Two
three-row evaluations through one dispatcher produced different columns: the
first began `6f155395-c8b9-436b-a39c-d247226bc2b2`, the second
`1188645d-be7b-4681-9936-8b24d7c51835`. Spark returned matching columns.
DataFusion 54 evaluates each projection expression separately, and the JVM
kernel cache receives no expression-occurrence ID. Output:
`/tmp/comet-4827-probe/result-final.log`.
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -847,6 +886,33 @@ object QueryPlanSerde extends Logging with CometExprShim
with CometTypeShim {
None
}
})
+ .orElse {
+ // Last-resort JVM expression detour. Any `None` above — no registered
handler, an
+ // `Unsupported`/opt-out `Incompatible` support level, or a handler
whose `convert`
+ // returned `None` — means this node has no native translation. Inside
a detour-eligible
+ // context (projection / filter, see `jvmDetourAllowed`) retry the
*outermost* unsupported
+ // node as a JVM detour: `emitJvmCodegenDispatch` ships its argument
columns over Arrow FFI
+ // and evaluates it with Spark's own codegen inside the Comet
pipeline, so the operator —
+ // and the native island around it — stays native. Supported ancestors
keep converting
+ // natively and reference this detour's output. If the detour cannot
fire (feature off,
+ // dispatcher off, or `CometBatchKernelCodegen.canHandle` rejects the
tree), `None`
+ // propagates unchanged and the operator falls back exactly as before.
+ if (jvmDetourAllowed) {
Review Comment:
[P2] Check bridge input support before rescuing interval expressions. With
both flags enabled, `SELECT abs(ym) FROM (SELECT make_ym_interval(CAST(rand(42)
* 10 AS INT)) AS ym FROM t)` over a native Parquet scan retains two projections
because the lower expression is nondeterministic. The new hook rescues both
`make_ym_interval` and interval `Abs`, which previously fell back. Although
`canHandle` accepts them, the lower projection produces an `IntervalYearVector`
that `CometScalaUDFCodegen.specFor` rejects, aborting the query instead of
returning Spark's interval results. Reject such inputs at planning time until
the dispatcher and generated input getters support them.
Evidence: The exact-head component harness verified Spark's two-project plan
and results `[P6Y], [P5Y], [P8Y]`. Both expressions passed `canHandle`.
Executing the lower expression produced `IntervalYearVector`; feeding that
actual output into the upper dispatcher reproduced
`UnsupportedOperationException: CometScalaUDFCodegen: unsupported Arrow vector
IntervalYearVector`. This was a bounded two-stage dispatcher reproduction
without native/JNI execution. Source and output are under
`/tmp/comet-4827-probe/`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
