peter-toth commented on PR #59522: URL: https://github.com/apache/spark/pull/59522#issuecomment-6340980826
Thanks for the concrete case, that helps. One note on the pre-4.2 behavior: when a key had more than one split, 4.1 grouped them and dropped the scan's reported ordering (`DataSourceV2ScanExecBase.outputOrdering`). So with 24 files per bucket, those scans had fewer tasks but no ordering before 4.2 either. Where a plan uses the scan's partitioning, like a storage-partitioned join or an aggregate on the bucket key, `EnsureRequirements` already groups the splits with a `GroupPartitionsExec`. So the scans that still run one task per split are the ones whose partitioning nothing uses: below a shuffle, or with no such consumer at all, like a collect, a write or the streamed side of a broadcast join. I'm not against a default-off option for this. Is the case below a shuffle the only one you need? If so, nothing there needs the splits grouped by key, since the shuffle redistributes the rows anyway. A coalesce placed where an exchange goes above the keyed scan would be enough, for example down to the scan's number of keys, which is the task count you had before 4.2. Placed only there, it can't break a consumer that relies on the keyed partitioning. Unlike grouping by key, it doesn't turn a key with many splits into one large task. It also wouldn't need the k-way merge handling, the `peel` and `withJoinKeyPositions` exceptions, or the `preserveScanGrouping` flag. If your connector reports `HasPartitionStatistics.sizeInBytes()`, the target could also be a size, like `spark.sql.files.maxPartitionBytes` for file sources. The current head also changes two plans with the option on: - With `spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled`, the SPJ is planned from the already grouped scan. On the SPARK-42038 test data the join runs 3 tasks instead of 7. - A `SORT BY id, name` over a source that reports `[id, name]` keeps its `SortExec`, also with `spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled`. Neither shape has an exchange above the scan, so a coalesce only there would leave both alone. Separately, SPARK-55092 changed this behavior in 4.2 without a migration note. If we agree on the option, could you add one in this PR? It could describe the 4.2 change and point to the option. -- 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]
