comphead commented on code in PR #5923:
URL: https://github.com/apache/datafusion-comet/pull/5923#discussion_r4277234938
##########
spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala:
##########
@@ -156,26 +156,30 @@ object CometBatchKernelCodegen extends Logging with
CometExprTraitShim with Come
// instance with a single `init(partitionIndex)` call, so `Rand` /
`MonotonicallyIncreasingID`
// state advances correctly across batches.
//
- // `ExecSubqueryExpression` (`ScalarSubquery`, `InSubqueryExec`) is
accepted: the surrounding
- // Comet operator's inherited `SparkPlan.waitForSubqueries` populates the
subquery's
- // `result` field before evaluation. The closure serializer captures that
value into the
- // arg-0 bytes, and the dispatcher keys its compile cache on those bytes,
so distinct subquery
- // results produce distinct cache entries.
+ // `ExecSubqueryExpression` (`ScalarSubquery`, `InSubqueryExec`) is
rejected. Its `result` field
+ // is populated by the surrounding operator's
`SparkPlan.waitForSubqueries` at *execution* time,
+ // whereas the dispatcher closure-serializes the tree into the arg-0 bytes
during planning. The
+ // deserialized copy on the executor therefore still has an unset
`result`, and
+ // `ScalarSubquery.doGenCode` fails the kernel compile with "Subquery ...
has not finished".
+ // Refusing here turns that runtime failure into a plan-time Spark
fallback. Note this only
+ // concerns a subquery *inside* the dispatched subtree: one that is merely
a sibling under a
+ // native parent (`udf(x) + (SELECT ...)`) is evaluated natively and never
reaches a kernel.
//
// `Unevaluable`: rejected by default. `isCodegenInertUnevaluable` exempts
version-specific
// leaves that are `Unevaluable` but never invoked by codegen (e.g. Spark
4.0's
// `ResolvedCollation` in `Collate.collation`, where `Collate.genCode`
delegates to its child).
boundExpr.find {
case _:
org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction => true
case _: org.apache.spark.sql.catalyst.expressions.Generator => true
+ case _: org.apache.spark.sql.execution.ExecSubqueryExpression => true
Review Comment:
`upstream/main` already rejects `ExecSubqueryExpression` in `canHandle`. It
came in with #6666 (commit `070951d2a7`), with the same explanation in the
comment about the closure-serialized tree and the "has not finished" failure,
so this hunk now conflicts and has nothing left to add. Could you rebase and
keep only what `main` lacks, which is the regression test and the reworded
comment on `ScalaUDF composed with reused scalar subquery across projection and
filter`? That comment is still on `main` and still describes the old
populate-before-serialize invariant. I only compared the two `canHandle` bodies
and did not run the new test against `main`'s fix.
--
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]
