ulysses-you commented on PR #59522: URL: https://github.com/apache/spark/pull/59522#issuecomment-6432136847
Thank you @peter-toth for the ping. V1 already had this problem, and it did not solve it with a task-count target. A bucketed `FileSourceScanExec` reads exactly `numBuckets` partitions and skips the size-based file packing (`maxPartitionBytes` / `openCostInBytes`) of the non-bucketed path. That only pays off when the bucket layout saves something, so `DisableUnnecessaryBucketedScan` runs right after `EnsureRequirements` and falls back to the default file-source parallelism when no operator with a `ClusteredDistribution` / `AllTuples` requirement is reachable, or when an `Exchange` separates the scan from it. `KeyGroupedPartitioning` is the V2 form of that claim. When SPJ cannot be used it buys nothing, yet it still shapes the plan. Today the only way to switch it off is the session-wide `spark.sql.sources.v2.bucketing.enabled`, which is all-or-nothing. I would add the V2 counterpart of `DisableUnnecessaryBucketedScan`: after `EnsureRequirements`, decide per query that no keyed distribution is consumed, and disable the keyed claim for that scan, so the scan's parallelism is whatever the connector's default planning produces, as it is for V1. Concretely I see two directions. Both are gated by a plan-level decision about what the consumers rely on; they differ in who performs the regrouping. **1. Disable the keyed layout when it buys nothing.** The timing is the awkward part: `Batch.planInputPartitions()` usually runs before `EnsureRequirements` has decided anything, so a Spark-side rule alone can withdraw the claim but cannot change the task count. The connector can re-plan, though, and the plan-level decision is the same for every connector. Today that decision has to be reimplemented in each connector; I did exactly that for Paimon in apache/paimon#3928, where `DisableUnnecessaryPaimonBucketedScan` is registered through a `SparkSessionExtensions` query-stage preparation rule, walks the finished physical plan with the same criteria as the V1 rule (an interesting `ClusteredDistribution` / `AllTuples` consumer reachable, or a `ShuffleExchangeLike` in between), and replaces the `BatchScanExec` with a copy whose scan has bucketed scan disabled. Because the scan is a copy rather than a mutation, partition planning runs again, and `getInputPartitions` falls back to the co nnector's default layout. That works, but every downstream connector would repeat the same traversal, the same criteria and the same gating, and the copies will drift apart. The general form is a small connector API instead: Spark owns the detection and triggers the non-bucketed path, and the connector only returns a scan variant that plans without bucketing, as a fresh object so the re-planning is well defined. A connector then reuses the non-bucketed planning path it already has for tables that are not bucketed, instead of reimplementing the rule. **2. Spark-side size-based re-sizing.** The gate here is not "SPJ did not happen". Coalescing inside a partition, or inside a key group, keeps the keyed layout intact, so the same sizing can apply when SPJ is in effect; only a regrouping that crosses partition boundaries requires the layout to be unused, which is the case 1 gates on. With that distinction, Spark can size the stage from an interface that already exists, and @peter-toth and @szehon-ho have both pointed at it: `HasPartitionStatistics` (`@since 4.0.0`, SPARK-42040) is a public mix-in on `InputPartition` that reports `sizeInBytes()`, `numRows()` and `filesCount()`, and its javadoc says Spark will use it to decide whether partition grouping applies, though nothing in Spark consumes it yet. Spark can sum the reported bytes, pick a target task size, and coalesce to `ceil(total / target)` tasks inside a partition, or across the stage when nothing relies on the layout, leaving the plan alone when sizes are absent. That is d ata-aware, unlike a key-count target, and it is not specific to keyed scans, so it could be the general option that was asked for here. Two limits: it only helps connectors that implement the mix-in (Iceberg does not today, and a connector like the one in this PR would need to report sizes), and coalescing only merges adjacent splits, so the resulting task sizes are approximate where a connector-side packer could do better. -- 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]
