rekbun opened a new pull request, #59522: URL: https://github.com/apache/spark/pull/59522
### What changes were proposed in this pull request? Add `spark.sql.sources.v2.bucketing.groupScanPartitions.enabled`, default `false`, to group scan splits sharing a reported partition key even when a downstream shuffle makes grouping unnecessary for correctness. - Insert `GroupPartitionsExec` above eligible V2 scans when the setting is enabled. - Mark grouping requested for scan task sizing and preserve it when `EnsureRequirements` removes grouping used for join alignment. - Preserve the marker through copying, canonicalization, and join-key projection. - Document the configuration and add plan and execution regressions. ### Why are the changes needed? [SPARK-60265](https://issues.apache.org/jira/browse/SPARK-60265). The SPJ refactor in [PR #54330](https://github.com/apache/spark/pull/54330) deliberately leaves scans ungrouped when a downstream shuffle makes grouping unnecessary (SPARK-55092). For V2 sources exposing many small physical splits per bucket, this can leave one scan task per split. A large number of scan tasks can consume significant scan-stage time through repeated task launch, setup, execution, and cleanup, increasing overall query latency. This patch provides explicit control over scan task sizing for these sources. The reproducer has two buckets with 24 physical splits each. An aggregation by a non-partition column requires a shuffle and executes 48 scan tasks by default. With the setting enabled, it executes two scan tasks, with identical query results. ### Does this PR introduce _any_ user-facing change? Yes: a new configuration, default `false`. It requires `spark.sql.sources.v2.bucketing.enabled` and usable reported partition keys surviving column pruning. Unique-key and unpartitioned scans are unaffected. ```sql SET spark.sql.sources.v2.bucketing.enabled=true; SET spark.sql.sources.v2.bucketing.groupScanPartitions.enabled=true; ``` Grouping can reduce scan-stage time by amortizing per-task costs across many small splits. Large or skewed keys can lose parallelism and produce longer tasks, so the behavior remains opt-in. Existing default scan behavior and the SPARK-55092 regression test are preserved. ### How was this patch tested? 46 focused tests passed: - Five `KeyGroupedPartitioningSuite` cases, including SPARK-55092, the bucketed aggregation and shuffled join, unique and unpartitioned scans, and V2 bucketing disabled. - One runtime-pruning case against each of the connector and Catalyst filter interfaces, including partially and fully pruned buckets. - Four `EnsureRequirementsSuite` cases covering grouping preservation, peeling, join-key projection, copying, and canonicalization. - All 35 `GroupPartitionsExecSuite` cases, including sorted-merge ordering coverage. The aggregation runs with AQE on and off. The join runs with AQE and the one-side-shuffle optimization independently on and off. The tests execute the scan-stage RDD and return one row count per task: 48 one-row tasks with grouping disabled, and two 24-row tasks with it enabled. Runtime pruning verifies task row counts `[0, 24]` and `[0, 0]` with correct query results. ```bash build/sbt \ 'sql/testOnly *KeyGroupedPartitioningSuite *KeyGroupedPartitioningRuntimeFilterSuite *KeyGroupedPartitioningCatalystRuntimeFilterSuite -- -z "Scans should not group partitions" -z "opt-in scan grouping"' \ 'sql/testOnly *EnsureRequirementsSuite -- -z "scan grouping" -z "peel" -z "withJoinKeyPositions"' \ 'sql/testOnly *GroupPartitionsExecSuite' build/sbt 'catalyst/scalastyle' 'sql/scalastyle' 'sql/Test/scalastyle' ``` Catalyst main, SQL main, and SQL test scalastyle checks passed. The task-count reduction was verified; wall-clock latency improvements have not been benchmarked. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Codex (GPT-6) -- 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]
