Neuw84 commented on issue #6133:
URL:
https://github.com/apache/datafusion-comet/issues/6133#issuecomment-5799048238
Caller stack captured for the q5 failure (Comet 1.0.0, Spark 4.1.3,
`CometNativeScan` with third-party operators):
```
SparkUnsupportedOperationException: SubqueryAdaptiveBroadcastExec does not
support the execute() code path.
at
org.apache.spark.sql.execution.SubqueryAdaptiveBroadcastExec.doExecute(SubqueryAdaptiveBroadcastExec.scala:44)
at
org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:378)
at
org.apache.spark.sql.execution.InSubqueryExec.updateResult(subquery.scala:134)
at
org.apache.spark.sql.comet.CometNativeScanExec.$anonfun$serializedPartitionData$1(CometNativeScanExec.scala:176)
at
org.apache.spark.sql.comet.CometNativeScanExec.serializedPartitionData$lzycompute(CometNativeScanExec.scala:171)
at
org.apache.spark.sql.comet.CometNativeScanExec.perPartitionData(CometNativeScanExec.scala:240)
at
org.apache.spark.sql.comet.CometNativeScanExec.outputPartitioning$lzycompute(CometNativeScanExec.scala:131)
at
org.apache.spark.sql.execution.exchange.ValidateRequirements$.validateInternal(ValidateRequirements.scala:48-50)
... (ValidateRequirements recursion, from
AdaptiveSparkPlanExec.optimizeQueryStage)
```
So the mechanism is the one the code comment in `serializedPartitionData`
describes: the inner `scan.partitionFilters` holds a **separate**
`InSubqueryExec` instance for the dynamic-pruning filter, which Spark's
expression walk does not see — so `PlanAdaptiveDynamicPruningFilters` rewrites
the outer copy but the inner one is still the placeholder
`SubqueryAdaptiveBroadcastExec` when `outputPartitioning` (reached by
`ValidateRequirements` inside `optimizeQueryStage`) forces
`serializedPartitionData`, which calls `updateResult()` on it → `execute()` →
this exception. The same eager, unsynchronised evaluation of the inner copy is
the natural explanation for the intermittent q64 zero rows: whatever the inner
`InSubqueryExec` resolves to at that moment is what prunes the file partitions,
and it is not the filter AQE ends up executing.
Why it shows with third-party operators and not in the pure Comet
configuration is presumably the plan shape `ValidateRequirements` walks (our
operators declare required distributions the walk checks down to the scan); the
root is independent of them. A fix on Comet's side would be to make the inner
copy follow the outer one (rewrite `scan.partitionFilters` when the wrapper's
`partitionFilters` are rewritten, or evaluate only the wrapper's
already-resolved `InSubqueryExec` values) and to keep `outputPartitioning` from
evaluating runtime subqueries at all (`originalPlan.outputPartitioning`, or a
count that does not depend on dynamic pruning).
--
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]
