ulysses-you opened a new pull request, #58681: URL: https://github.com/apache/spark/pull/58681
### What changes were proposed in this pull request? Turn three Storage Partition Join (SPJ) configurations on by default, and document the whole `spark.sql.sources.v2.bucketing.*` family, whose remaining members were partly undocumented (`spark.sql.sources.v2.bucketing.sorting.enabled` and `spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled` were missing from the performance tuning guide). The table below covers every configuration in the family, its default after this change, and why the ones that are not flipped keep their current value. | Configuration | Default | Why | |---|---|---| | `spark.sql.sources.v2.bucketing.enabled` | `true` (unchanged) | Master switch for recognizing a V2 source's reported partitioning. | | `spark.sql.sources.v2.bucketing.pushPartValues.enabled` | `true` (unchanged) | Pushes the merged partition key list down so both join legs align. A prerequisite for the flags below. | | `spark.sql.sources.v2.bucketing.partition.filter.enabled` | **`true` (this change)** | Narrows the key groups pushed down to both join sides to those that can produce output for the join type (intersection for inner/semi joins, one side for the variants driven by that side, union for full outer). A dropped group is never scanned, so this is strictly less work and needs no statistics. | | `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` | **`true` (this change)** | A scan that reports a keyed partitioning but no explicit ordering is constant on its partition key expressions within each partition, so it can report that ordering. Planning-time only, and it rests on the same `HasPartitionKey` contract that eliminating the shuffle already rests on. | | `spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled` | **`true` (this change)** | When `GroupPartitionsExec` merges partitions sharing one key value, orders over the partition key expressions still hold. Report-only: `doExecute` is identical either way, and the orders it keeps are over expressions that are constant in the output partition. | | `spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled` | `false` (kept) | Merges by a k-way sorted merge to keep the child's *full* ordering. Real cost: per-row comparisons and a priority queue instead of concatenation, and it gives up columnar execution (`supportsColumnar` is false when sorted merge is used). Beneficial only when the data is both partitioned and sorted, so it stays an opt-in. | | `spark.sql.sources.v2.bucketing.sorting.enabled` | `false` (kept) | Removes the range shuffle for a sort on the partition keys, but that shuffle is what balances the sorted output: parallelism and skew handling then belong to the data source's split layout, and adaptive coalescing and skew splitting lose the stage to work on. A few large splits become a few large tasks. | | `spark.sql.sources.v2.bucketing.shuffle.enabled` | `false` (kept) | One-sided shuffle onto a keyed layout costs more than it looks: the keyed spec is what disables the minimum-parallelism floor, so a small-bucket dimension can pull a large fact table down to a few partitions; the shuffled side computes its partition with interpreted `Expression.eval` per row instead of a codegen'd projection; the key-to-partition map is built on the driver and serialized into every task; and the resulting stage is invisible to AQE. | | `spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled` | `false` (kept) | Groups and replicates one side to match the other, choosing the big table from table statistics. Without reliable statistics this can replicate the wrong side and inflate memory, which is the failure mode it was gated for. | | `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` | `false` (kept) | Groups on a subset of the partition keys (and on keys that collapsed onto each other), which produces partitions larger than the source declared: coarser grouping means skew and less parallelism, exactly what the configuration is documented to expose. | | `spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled` | `false` (kept) | Reduces compatible-but-different transforms onto a common key space, so `GroupPartitionsExec` merges splits the source kept apart (`hours` against `days` turns 24 groups into one) with no statistics consulted, gives up key-derived ordering, blocks downstream reuse of the partitioning, and can raise `STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES` at planning time for a connector whose reducer returns another type. It is also mutually exclusive with partially clustered distribution. | Two related configurations are outside the `v2.bucketing` namespace and are untouched: `spark.sql.requireAllClusterKeysForCoPartition` stays `true` (it gates a V2 storage-partitioned join on every join key being covered by a partition key, and setting it `false` is what exposes the skew above), and `spark.sql.requireAllClusterKeysForDistribution` already defaults to `false`. ### Why are the changes needed? The three flipped defaults are the SPJ members that only remove work: the first scans a strict subset of what it scans today, the other two report orderings that already hold and let Spark drop a `Sort` it would otherwise add. None of them adds runtime cost, moves or replicates data, or consults table statistics, so leaving them off costs users the optimization without buying any safety. Every configuration that is kept off trades shuffle elimination for skew, extra per-row cost, or memory, as the table spells out. ### Does this PR introduce _any_ user-facing change? Yes, within unreleased branches. Three configurations now default to `true`. They apply only when `spark.sql.sources.v2.bucketing.enabled` is on, and partition filtering additionally requires `spark.sql.sources.v2.bucketing.pushPartValues.enabled`, both already `true`. A physical plan may now show fewer `Sort` nodes above a V2 scan and above `GroupPartitionsExec`, and an SPJ may read fewer files because key groups that cannot match are dropped instead of padded with empty partitions. Set any of the three to `false` to restore the previous behavior; the SQL migration guide records the same. ### How was this patch tested? `build/sbt 'sql/testOnly org.apache.spark.sql.connector.*'` (75 suites, 4045 tests), plus `EnsureRequirementsSuite`, `ValidateRequirementsSuite`, `ProjectedOrderingAndPartitioningSuite`, `ShuffleSpecSuite`, `DistributionSuite` and `GroupPartitionsExecSuite`. Cases that pinned the old default now assert the new one with the off branch kept explicit (the `SPARK-56241` ordering tests in `GroupPartitionsExecSuite` and `KeyGroupedPartitioningSuite`). Cases that assert the union of both sides' partition keys, where union padding is not what they exist to check, now set `spark.sql.sources.v2.bucketing.partition.filter.enabled` to `false` explicitly (`SPARK-42038`, `SPARK-48065`, `SPARK-55535`, `SPARK-55992`, `SPARK-56182`, `SPARK-57881`, `SPARK-59050`). The scalastyle tasks for `catalyst` and `sql` are clean. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
