[
https://issues.apache.org/jira/browse/SPARK-60265?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rakesh Kumar updated SPARK-60265:
---------------------------------
Component/s: SQL
(was: Optimizer)
Description:
Add a default-off option that packs DataSource V2 batch scan inputs by
{{HasPartitionStatistics.sizeInBytes()}}, with an opening cost per input and an
optional count cap, when no operator uses the scan's partitioning or ordering.
This gives DSv2 scans the size-based task sizing that file sources have with
{{spark.sql.files.maxPartitionBytes}}, and addresses the extra scan tasks that
connectors with many small splits per key see after SPARK-55092.
New configurations: {{spark.sql.sources.v2.scanPartitionPacking.enabled}}
(default false), {{.maxPartitionBytes}}, {{.openCostInBytes}},
{{.minPartitionNum}} and {{.maxPartitionNum}}.
A separate rule after {{EnsureRequirements}} packs a scan only below a shuffle
or broadcast exchange, on the streamed side of a broadcast join, or where it
has no consumer. Cached plans and checkpoints are not packed. Pull request:
https://github.com/apache/spark/pull/59522
was:
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.
Summary: Add optional size-based packing for DataSource V2 batch scans
(was: Allow optional grouping of V2 scan splits by partition key to reduce
tasks bloat below shuffles.)
> Add optional size-based packing for DataSource V2 batch scans
> -------------------------------------------------------------
>
> Key: SPARK-60265
> URL: https://issues.apache.org/jira/browse/SPARK-60265
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 4.2.0
> Reporter: Rakesh Kumar
> Priority: Major
> Labels: pull-request-available
>
> Add a default-off option that packs DataSource V2 batch scan inputs by
> {{HasPartitionStatistics.sizeInBytes()}}, with an opening cost per input and
> an optional count cap, when no operator uses the scan's partitioning or
> ordering. This gives DSv2 scans the size-based task sizing that file sources
> have with {{spark.sql.files.maxPartitionBytes}}, and addresses the extra scan
> tasks that connectors with many small splits per key see after SPARK-55092.
> New configurations: {{spark.sql.sources.v2.scanPartitionPacking.enabled}}
> (default false), {{.maxPartitionBytes}}, {{.openCostInBytes}},
> {{.minPartitionNum}} and {{.maxPartitionNum}}.
> A separate rule after {{EnsureRequirements}} packs a scan only below a
> shuffle or broadcast exchange, on the streamed side of a broadcast join, or
> where it has no consumer. Cached plans and checkpoints are not packed. Pull
> request: https://github.com/apache/spark/pull/59522
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
