[
https://issues.apache.org/jira/browse/SPARK-60265?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-60265:
-----------------------------------
Labels: pull-request-available (was: )
> Allow optional grouping of V2 scan splits by partition key to reduce tasks
> bloat below shuffles.
> ------------------------------------------------------------------------------------------------
>
> Key: SPARK-60265
> URL: https://issues.apache.org/jira/browse/SPARK-60265
> Project: Spark
> Issue Type: Improvement
> Components: Optimizer
> Affects Versions: 4.2.0
> Reporter: Rakesh Kumar
> Priority: Major
> Labels: pull-request-available
>
> Problem Statement
> The SPJ refactor in PR #54330 (https://github.com/apache/spark/pull/54330)
> moved partition grouping from BatchScanExec to GroupPartitionsExec. Grouping
> is omitted when a downstream shuffle makes it unnecessary for correctness, as
> intended by SPARK-55092 (https://issues.apache.org/jira/browse/SPARK-55092).
> For DataSource V2 connectors exposing many small physical splits per
> partition key, this can produce one scan task per split and increase
> scheduling overhead.
> Proposed change
> Add an opt-in configuration:
> spark.sql.sources.v2.bucketing.groupScanPartitions.enabled
> Default: false.
> When enabled:
> • Group scan splits sharing a reported partition key using
> GroupPartitionsExec.
> • Preserve explicitly requested scan grouping through EnsureRequirements,
> including when a downstream shuffle is required.
> • Require V2 bucketing and usable partition keys surviving column pruning.
> • Leave unique-key and unpartitioned scans unaffected.
> Preserve the existing default behavior and SPARK-55092 regression test.
> Reproducer and validation
> A local reproducer with two buckets and 24 physical splits per bucket
> executes 48 scan tasks by default. The proposed setting reduces this to 2
> scan tasks, with identical results.
> The prototype passed 46 focused tests covering AQE, shuffled joins, runtime
> filtering, empty buckets, grouping preservation, and existing grouping
> behavior. Scalastyle checks also passed.
> Tradeoff
> Grouping can reduce scheduling overhead for small splits, but large or skewed
> buckets can lose parallelism. The setting should remain opt-in.
> Representative workload benchmarking is needed to establish a wall-clock
> speedup.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
