You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "xhumanoid (via GitHub)" <gi...@apache.org> on 2026/09/14 15:27:55 UTC
[PR] fix: reject scalar subqueries inside a codegen-dispatch kernel [datafusion-comet]
xhumanoid opened a new pull request, #5923:
URL: https://github.com/apache/datafusion-comet/pull/5923
A subquery inside a dispatched subtree fails at runtime. `SELECT addOne((SELECT max(v) FROM t2)) FROM t` dies with "Subquery ... has not finished" from ScalarSubquery.doGenCode on the executor, after the plan has already committed to the kernel and can no longer fall back.
canHandle claimed to support ExecSubqueryExpression on the grounds that the surrounding operator's waitForSubqueries populates the subquery's `result` before the dispatcher serializes the expression. That ordering never held: the dispatcher closure-serializes the tree during planning, while `result` is populated at execution time, so the deserialized copy on the executor always has an unset result. The existing reused-subquery test passes only because its subqueries sit under native parents (`addOne(x) + (SELECT ...)`), where they are evaluated natively and never reach a kernel.
canHandle now refuses them, which turns the runtime failure into a plan-time Spark fallback, and the comment is corrected to say so.
## Which issue does this PR close?
Closes #.
## How are these changes tested?
add regression test
failed without 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: reject scalar subqueries inside a codegen-dispatch kernel [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: reject scalar subqueries inside a codegen-dispatch kernel [datafusion-comet]
Posted by "xhumanoid (via GitHub)" <gi...@apache.org>.
xhumanoid commented on PR #5923:
URL: https://github.com/apache/datafusion-comet/pull/5923#issuecomment-5666815555
@mbutrovich ok, will check
thanks for pointing one more place
found this bug during investigation of HiveSimpleUDF support
will create pr separately, started from Spark 3.5 they have doCodeGen method
https://issues.apache.org/jira/browse/SPARK-42051
https://issues.apache.org/jira/browse/SPARK-42052
I used ScalaUDF as a base harness and catch it on some of my tests
main different with existing unit test:
existing do evaluation of subquery separately and ScalaUDF independent, it's not a part of UDF kernel:
```sql
SELECT addOne(x) + (SELECT max(v) FROM t2) AS r
FROM t
WHERE addOne(x) < (SELECT max(v) FROM t2) * 2
```
in my case subquery is a part of UDF parameter
```sql
SELECT addOne( (SELECT max(v) FROM t2) ) AS r
FROM t
```
so during code generation we can't use the value returned from subquery
without patch exception from driver
```
Caused by: java.lang.IllegalArgumentException: requirement failed: Subquery subquery#2254, [id=#3678] has not finished
at scala.Predef$.require(Predef.scala:337)
at org.apache.spark.sql.execution.ScalarSubquery.toLiteral(subquery.scala:108)
at org.apache.spark.sql.execution.ScalarSubquery.doGenCode(subquery.scala:104)
at org.apache.spark.sql.catalyst.expressions.Expression.$anonfun$genCode$3(Expression.scala:229)
at scala.Option.getOrElse(Option.scala:201)
at org.apache.spark.sql.catalyst.expressions.Expression.genCode(Expression.scala:224)
at org.apache.spark.sql.catalyst.expressions.KnownNotNull.doGenCode(constraintExpressions.scala:48)
at org.apache.spark.sql.catalyst.expressions.Expression.$anonfun$genCode$3(Expression.scala:229)
at scala.Option.getOrElse(Option.scala:201)
at org.apache.spark.sql.catalyst.expressions.Expression.genCode(Expression.scala:224)
at org.apache.spark.sql.catalyst.expressions.ScalaUDF.$anonfun$doGenCode$2(ScalaUDF.scala:1118)
at scala.collection.immutable.ArraySeq.map(ArraySeq.scala:75)
at scala.collection.immutable.ArraySeq.map(ArraySeq.scala:35)
at org.apache.spark.sql.catalyst.expressions.ScalaUDF.doGenCode(ScalaUDF.scala:1118)
at org.apache.spark.sql.catalyst.expressions.Expression.$anonfun$genCode$3(Expression.scala:229)
at scala.Option.getOrElse(Option.scala:201)
```
query itself still reused as expected
```
+- CometProject [r#2255], [if (isnull(Subquery subquery#2254, [id=#3677])) null else addOne(knownnotnull(ReusedSubquery Subquery subquery#2254, [id=#3677])) AS r#2255]
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: reject scalar subqueries inside a codegen-dispatch kernel [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #5923:
URL: https://github.com/apache/datafusion-comet/pull/5923#issuecomment-5666510814
Thanks for looking at this @xhumanoid. Could we get a deeper investigation into this? I didn't hallucinate that documentation that scalar subqueries work in a codegen'd expression. We have a test for it, but it's specifically for a reused subquery. I wonder if that changes the execution semantics somehow.
https://github.com/apache/datafusion-comet/blob/481aefea9c60592612650de73bd2e1c7aa173979/spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala#L1146
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: reject scalar subqueries inside a codegen-dispatch kernel [datafusion-comet]
Posted by "xhumanoid (via GitHub)" <gi...@apache.org>.
xhumanoid commented on PR #5923:
URL: https://github.com/apache/datafusion-comet/pull/5923#issuecomment-5710165378
@mbutrovich ping
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: reject scalar subqueries inside a codegen-dispatch kernel [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #5923:
URL: https://github.com/apache/datafusion-comet/pull/5923#discussion_r4155998248
##########
spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala:
##########
@@ -1143,6 +1143,24 @@ class CometCodegenSuite
checkSparkAnswerAndOperator(df)
}
+ test("a scalar subquery inside the dispatched subtree falls back to Spark") {
+ // The dispatcher serializes the tree at plan time, but a subquery's `result` is populated by
+ // the surrounding operator's `waitForSubqueries` at execution time. The deserialized copy on
+ // the executor therefore has no result, and `ScalarSubquery.doGenCode` used to fail the kernel
+ // compile with "Subquery ... has not finished" -- a runtime failure, after the plan had already
+ // committed to the kernel. `canHandle` now refuses it, so the operator falls back cleanly.
+ spark.udf.register("addOne", (i: Int) => i + 1)
+ withTable("t", "t2") {
+ sql("CREATE TABLE t (x INT) USING parquet")
+ sql("INSERT INTO t VALUES (1), (2), (3), (4), (5)")
+ sql("CREATE TABLE t2 (v INT) USING parquet")
+ sql("INSERT INTO t2 VALUES (2), (4)")
+ assertNoCodegen {
+ checkSparkAnswer(sql("SELECT addOne((SELECT max(v) FROM t2)) AS r FROM t"))
+ }
+ }
+ }
+
test("ScalaUDF composed with reused scalar subquery across projection and filter") {
Review Comment:
The comment on this existing test now contradicts the fix. It says `waitForSubqueries` populates each subquery's `result` before the bridge serializes the expression, that the value travels into the cache key bytes so different subquery values compile distinct kernels, and that the test exercises the "subquery-correctness invariant documented on `canHandle`". The new test and the updated `canHandle` comment show that ordering never held, and that invariant no longer exists. Could we update this comment to say what the test covers now? As I read it, that is a subquery under a native parent that is evaluated natively and never reaches a kernel.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: reject scalar subqueries inside a codegen-dispatch kernel [datafusion-comet]
Posted by "xhumanoid (via GitHub)" <gi...@apache.org>.
xhumanoid commented on code in PR #5923:
URL: https://github.com/apache/datafusion-comet/pull/5923#discussion_r4163379079
##########
spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala:
##########
@@ -1143,6 +1143,24 @@ class CometCodegenSuite
checkSparkAnswerAndOperator(df)
}
+ test("a scalar subquery inside the dispatched subtree falls back to Spark") {
+ // The dispatcher serializes the tree at plan time, but a subquery's `result` is populated by
+ // the surrounding operator's `waitForSubqueries` at execution time. The deserialized copy on
+ // the executor therefore has no result, and `ScalarSubquery.doGenCode` used to fail the kernel
+ // compile with "Subquery ... has not finished" -- a runtime failure, after the plan had already
+ // committed to the kernel. `canHandle` now refuses it, so the operator falls back cleanly.
+ spark.udf.register("addOne", (i: Int) => i + 1)
+ withTable("t", "t2") {
+ sql("CREATE TABLE t (x INT) USING parquet")
+ sql("INSERT INTO t VALUES (1), (2), (3), (4), (5)")
+ sql("CREATE TABLE t2 (v INT) USING parquet")
+ sql("INSERT INTO t2 VALUES (2), (4)")
+ assertNoCodegen {
+ checkSparkAnswer(sql("SELECT addOne((SELECT max(v) FROM t2)) AS r FROM t"))
+ }
+ }
+ }
+
test("ScalaUDF composed with reused scalar subquery across projection and filter") {
Review Comment:
sure, will update it
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: reject scalar subqueries inside a codegen-dispatch kernel [datafusion-comet]
Posted by "xhumanoid (via GitHub)" <gi...@apache.org>.
xhumanoid commented on PR #5923:
URL: https://github.com/apache/datafusion-comet/pull/5923#issuecomment-5667442020
```sql
SELECT addOne( (SELECT 123) ) AS r
FROM t
```
works as expected without patch, so i will keep investigation
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org