Re: [I] CometNativeScanExec evaluates the inner (unrewritten) dynamic-pruning subquery from outputPartitioning: q5 fails with SubqueryAdaptiveBroadcastExec.execute(), q64 intermittently returns 0 rows at SF1000 [datafusion-comet]

["Neuw84 (via GitHub)" <gi...@apache.org>]

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: 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