You are viewing a plain text version of this content. The canonical link for it is here.

Posted to github@datafusion.apache.org by "dwsmith1983 (via GitHub)" <gi...@apache.org> on 2026/09/27 15:11:05 UTC

[PR] fix: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

dwsmith1983 opened a new pull request, #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268

## Which issue does this PR close?

Part of #6133. This fixes the q5 failure. The q64 row loss has a separate cause, tracked in #6264.

## Rationale for this change

AQE applies Spark's own query stage rules before Comet's. When `CoalesceShufflePartitions` changes a stage, `optimizeQueryStage` runs `ValidateRequirements`, which reads `outputPartitioning` on every node in the stage. `CometPlanAdaptiveDynamicPruningFilters` has not run yet at that point, so a scan's DPP subquery is still Spark's `SubqueryAdaptiveBroadcastExec` placeholder.

For a non-bucketed scan, `CometNativeScanExec.outputPartitioning` returned `UnknownPartitioning(perPartitionData.length)`. Computing that length runs `InSubqueryExec.updateResult` on the placeholder, which throws `SubqueryAdaptiveBroadcastExec does not support the execute() code path`. That is the stack trace in the issue.

Coalescing skips shuffle reads that share a subtree with a scan, and `CoalesceShufflePartitions` only splits a stage into separate groups at Spark's `UnionExec`, `BroadcastHashJoinExec`, `BroadcastNestedLoopJoinExec` and `CartesianProductExec`. So the failure needs one of those in the scan's stage. q5 has a union and hits it when Comet's union is not used.

`FileSourceScanExec` does not have this problem: it reports `UnknownPartitioning(0)` for a non-bucketed scan and only resolves DPP when it builds its RDD.

## What changes are included in this PR?

- `CometNativeScanExec.outputPartitioning` reports `UnknownPartitioning(0)` for a non-bucketed scan, as `FileSourceScanExec` does. A bucketed scan still reports its bucket layout.
- `CometNativeExec` sizes the native RDD of a block whose leaf is a `CometScanWithPlanData` from `perPartitionData.length`, at execution, after `findAllPlanData` has resolved the scan's subqueries. It used to read that count from `outputPartitioning`. A bucketed scan has one entry per bucket, so its count is unchanged.

Planning now sees 0 partitions for a non-bucketed native scan, the same as for Spark's scan. In particular a single split scan no longer satisfies `AllTuples` while planning, which also matches Spark. The execution paths that need a count take it from the RDD.

## How are these changes tested?

Two tests in `CometExecSuite`, next to the existing AQE DPP tests:

- "AQE DPP: shuffle coalescing in the scan's stage keeps DPP working" joins a partitioned fact table to a filtered dimension and unions it with an aggregate, with Comet's union disabled so Spark's `UnionExec` keeps the aggregate's shuffle read in the scan's stage. On main it fails with the stack trace from the issue. With the change it matches Spark, and on Spark 3.5 and later it checks that the DPP scan and a coalesced shuffle read share the final stage, and that DPP still prunes to 1 of 10 partitions.
- "AQE DPP: bucketed scan keeps its partitioning for a sort-merge join" checks that a bucketed scan with DPP still feeds a sort-merge join without a shuffle. It passes on main and guards the bucketed branch.

`CometExecSuite`, `CometScanWithPlanDataSuite`, `CometTopKSuite`, `CometJoinSuite` and the DPP fallback suites pass on Spark 3.5. The AQE DPP tests and `CometTopKSuite` pass on Spark 4.0 and 4.1.

--
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: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.

dwsmith1983 commented on PR #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268#issuecomment-5907408435

   @sunchao could you approve a CI run on `588c029f7`? CI has not run on this PR yet, and it fixes the q5 failure from #6133.


-- 
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: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.

andygrove merged PR #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268


-- 
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: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.

dwsmith1983 commented on PR #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268#issuecomment-6095042988

   @andygrove the `CometTaskMetricsSuite` failure that took this out of the merge queue is fixed, and the branch is synced with `main` at `ab074fc6b`. The CI run there needs an approval before it can go back in the queue.
   


-- 
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: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.

dwsmith1983 commented on PR #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268#issuecomment-6007395851

   The failure on every profile was `CometTaskMetricsSuite` "coalesced native scan reports rows and bytes from every partition it reads", which #6547 added after this branch last merged main. Its precondition read `outputPartitioning.numPartitions` on the executed scan and expected 4, and this PR makes a non-bucketed scan report `UnknownPartitioning(0)`. e4bdd8024 has the test read `perPartitionData.length`, the count the native RDD is sized from, which is what the check meant. The suite passes with the change on the Spark 4.1 profile.
   


-- 
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: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.

andygrove commented on code in PR #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268#discussion_r4185303224


##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1318,6 +1325,9 @@ abstract class CometLeafExec extends CometNativeExec with LeafExecNode {
  * parent's native execution receives an empty input. (`CometIcebergNativeScanExec` does NOT use
  * this trait; it has a dedicated `findAllPlanData` case.)
  *
+ * `perPartitionData.length` is the partition count the native block runs with, and the count in

Review Comment:
   Now that `buildNativeContext` sizes every `CometScanWithPlanData` from `perPartitionData`, could this doc also say that `outputPartitioning` must not read `perPartitionData`, or anything else that runs the scan's DPP subqueries? AQE reads it while optimizing a stage, before `CometPlanAdaptiveDynamicPruningFilters` has converted the placeholders, and that's the q5 crash. The Delta scan in #5365 copied the old pattern and works around it with `hasUnevaluableSubqueryFilter`. With the rule written here it could just report `UnknownPartitioning(0)` like `CometNativeScanExec` does now.



-- 
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: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.

dwsmith1983 commented on code in PR #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268#discussion_r4186652194


##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1318,6 +1325,9 @@ abstract class CometLeafExec extends CometNativeExec with LeafExecNode {
  * parent's native execution receives an empty input. (`CometIcebergNativeScanExec` does NOT use
  * this trait; it has a dedicated `findAllPlanData` case.)
  *
+ * `perPartitionData.length` is the partition count the native block runs with, and the count in

Review Comment:
   > could this doc also say that `outputPartitioning` must not read `perPartitionData`, or anything else that runs the scan's DPP subqueries?
   
   Added in 2eb4ff65a. The `CometScanWithPlanData` doc now says `perPartitionData.length` is what sizes the native RDD at execution, that `outputPartitioning` must not read `perPartitionData` or anything that evaluates the scan's DPP subqueries because AQE calls it while optimizing a stage before `CometPlanAdaptiveDynamicPruningFilters` has converted the placeholders, and that a scan that cannot report a real partitioning without them should report `UnknownPartitioning(0)`, as `CometNativeScanExec` does for a non-bucketed scan.
   
   I pushed this while the label runs were still going so the doc change would not wait on them. It only touches that comment, so the code those runs are testing is unchanged.
   



-- 
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: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

Posted by "Neuw84 (via GitHub)" <gi...@apache.org>.

Neuw84 commented on PR #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268#issuecomment-5913466701

We tested this PR together with #6270 on TPC-DS at SF1000: Parquet on S3, a Spark 4.1 cluster on EKS, 8 executors x 13 cores on m5.4xlarge, one availability zone, AQE on with 300 shuffle partitions.

The build was main `b58b2f3a` with #6268 (`588c029f`) and #6270 (`b3f4f058`) merged on top (HEAD `0f22d064`), with the native library built for x86-64-v3. Comet ran with native scan, exec and shuffle enabled.

- **q5 now completes.** 23.1 s, 100 rows, the same checksum as vanilla Spark, with 77/77 operators on Comet (11x `CometNativeScanExec`). Before this fix, q5 failed at this scale with native scans.
- **The whole run passes.** All 103 queries complete. Row counts match vanilla Spark on every query, and checksums match on all but q65, which has ties in its ORDER BY and differs the same way for every engine we compare.

Thanks for the 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

Re: [PR] fix: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.

andygrove commented on PR #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268#issuecomment-5955547328

   I would like to do a deeper review of this PR before it is merged.


-- 
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: do not resolve DPP subqueries when AQE reads a native scan's partitioning [datafusion-comet]

Posted by "dwsmith1983 (via GitHub)" <gi...@apache.org>.

dwsmith1983 commented on PR #6268:
URL: https://github.com/apache/datafusion-comet/pull/6268#issuecomment-5861442511

   @andygrove could you approve a CI run on `a88941e2e`? CI has not run on this PR yet. It now also covers the native Iceberg scan, which failed the same way.
   


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