You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ulysses-you (via GitHub)" <gi...@apache.org> on 2026/09/10 03:28:43 UTC
[PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r3995045887
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2542,12 +2542,13 @@ object SQLConf {
buildConf("spark.sql.sources.v2.bucketing.partition.filter.enabled")
.doc(s"Whether to filter partitions when running storage-partition join. " +
s"When enabled, partitions without matches on the other side can be omitted for " +
- s"scanning, if allowed by the join type. This config requires both " +
- s"${V2_BUCKETING_ENABLED.key} and ${V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key} to be " +
- s"enabled.")
+ s"scanning, if allowed by the join type. This config requires " +
+ s"${V2_BUCKETING_ENABLED.key} to be enabled, together with either " +
+ s"${V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key} or " +
+ s"${V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key}.")
.version("4.0.0")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
val V2_BUCKETING_SORTING_ENABLED =
buildConf("spark.sql.sources.v2.bucketing.sorting.enabled")
Review Comment:
**Non-blocking (P2):** This default flip makes the existing SPARK-59272 path reachable with only `v2.bucketing.shuffle.enabled` opted in. If filtering intersects a marked side below its declared partition count, `GroupPartitionsExec` reports `UnknownPartitioning` after `EnsureRequirements` has already committed the pair; `ValidateRequirements` then rejects AQE shuffle-read and skew rewrites for that stage. Since the prerequisite fix is still open, please keep the agreed landing order or guard this default-on path so the selected children continue to satisfy the join distributions.
**Recommended change:** Make partition-filtered alignment accept a selected pair only when each resulting child still satisfies its required distribution; otherwise fall back to the existing safe, unfiltered alignment/shuffle path instead of committing the marked GroupPartitionsExec plan.
**Why this works:** Evaluate distribution satisfaction after the filtering and grouping transforms have produced their effective output partitioning. If a marked side would take the GroupPartitionsExec give-up path and expose UnknownPartitioning, discard that filtered candidate and continue through the established non-filtered requirement-enforcement path. Keep filtering enabled for candidates whose effective partitioning remains valid.
**Scope:** sql/core/src/main/scala/org/apache/spark/sql/execution/exchange, sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2, sql/core/src/test/scala/org/apache/spark/sql/connector, sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2
**Compatibility:** Keep correct query results, valid non-marked partition filtering, and the documented false opt-out unchanged.
**Risks:** The fallback can introduce a shuffle or retain more partitions for the affected marked case, reducing the optimization benefit. A validation placed before the effective GroupPartitionsExec output is known could repeat the current false-positive selection.
**Constraints:** Do not change query answers or permit a child that fails its join distribution requirement. Do not disable partition filtering for unmarked or otherwise distribution-valid pairings. Preserve the explicit false rollback behavior of the configuration.
**Success:** Every shuffle-free join pair selected after partition filtering satisfies both required distributions. A marked side whose filtered grouping would report UnknownPartitioning uses a valid fallback plan. Distribution-valid partition-filtered joins retain their existing pruning and shuffle-avoidance behavior.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1779,19 +1779,24 @@ class EnsureRequirementsSuite extends SharedSparkSession {
val right = new DummySparkPlanWithBatchScanChild(
outputPartitioning = KeyedPartitioning(Seq(days(aR), years(bR)), rightKeys))
- // No `withSQLConf` on purpose. `pushPartValues` is on by default, which is all the push branch
- // needs, so this is what a user gets out of the box.
- val smj = SortMergeJoinExec(Seq(xL, yL), Seq(aR, bR), Inner, None, left, right)
- val planned = EnsureRequirements.apply(smj)
-
- // Without the pairing, the left's first member is taken and its `days(yL)` is matched against
- // `days(aR)`, which names the other join key, so the join declines and both sides are shuffled
- // onto the default partitioning.
- assert(planned.collect { case s: ShuffleExchangeExec => s }.isEmpty,
- "the second member pairs with the other side, so neither side is shuffled")
- assert(groupPartitionsNodes(planned).map(_.expectedPartitionKeys.map(_.size)) ===
- Seq(Some(3), Some(3)),
- "both sides are pushed the union of the two key sets")
+ // `pushPartValues` is left at its default (on), which is all the push branch needs, so this is
+ // what a user gets out of the box. `partition.filter` is varied to cover both settings: off
+ // pushes the union of the two key sets, on (the default) their intersection.
+ Seq(3 -> false, 1 -> true).foreach { case (expected, filter) =>
+ withSQLConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> filter.toString) {
Review Comment:
**Non-blocking (P2):** This arm is described as covering the default, but `withSQLConf` explicitly sets `V2_BUCKETING_PARTITION_FILTER_ENABLED`. The other changed partition-filter cases also supply `true` or `false`, and SPARK-59050 pins `false`, so reverting `createWithDefault(true)` would leave the focused tests green. Could one distribution-valid planning case leave this setting unset and assert the intended enabled plan/result, while keeping an explicit `false` rollback case?
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.sorting.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, Spark satisfies a sort on the partition key expressions from the partitioning reported by a V2 data source, so no shuffle is added for that sort. The parallelism of the sorted output is then whatever the data source's partition layout provides, and there is no shuffle stage left for adaptive partition coalescing or skew splitting to balance. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> merges partitions that share a key by a sorted merge rather than by concatenation, so it can report the child's full ordering instead of only the orderings over partition key expressions that <code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code> preserves. This removes a downstream sort when data is both partitioned and sorted, but a sorted merge costs more than concatenation, especially when merging many partitions, and it gives up columnar execution for the merged plan. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
Review Comment:
**Non-blocking (P2):** Enabling this setting does not by itself make `GroupPartitionsExec` use sorted merge. `EnsureRequirements` only tries it when a required ordering is otherwise unsatisfied, and the node additionally requires coalescing, non-empty child ordering, and a `SafeForKWayMerge` subtree; otherwise it still concatenates and does not report the full child ordering. Could this say sorted merge *may* be selected when those demand and feasibility conditions hold?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r4034344416
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -8191,17 +8234,27 @@ class KeyGroupedPartitioningSuite
withSQLConf(
SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
"spark.sql.autoBroadcastJoinThreshold" -> "-1",
- SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ // Filtering narrows the joined groups to the intersection, which leaves 2 groups over
+ // the marked side's 3 input partitions. The grouping stays index-identity but forfeits
+ // its keyed claim (SPARK-59272), and this test pins that claim surviving a regrouping,
+ // so it runs with filtering off.
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "false") {
Review Comment:
**Finding 6.** This pin and the count clause you added to `identityGpe` are both obsolete once you rebase, and the conflict here is where you will meet that.
#58858 rewrote this block: the test now runs both settings of `partition.filter.enabled`, asserts `ValidateRequirements` and asserts `k.numPartitions === 3`, so the marked side keeping all three declared keys is pinned directly. Take master's side of the hunk and both the pin and the tightening go with it.
On the tightening specifically, so it does not look like it was dropped silently: it was your answer to finding 1 last round and it was right, but master's finder no longer needs it. The comment there says why — the finder is deliberately weaker than `identityGrouping`, and `numPartitions === 3` is what carries the assertion. Nothing you fixed is lost.
Resolve the hunk, not the file. Taking master's whole `KeyGroupedPartitioningSuite.scala` also drops your `SPARK-55992` pin and all seven parameterized cases — I did exactly that on my first attempt before noticing.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5235,7 +5259,9 @@ class KeyGroupedPartitioningSuite
sql(s"INSERT INTO testcat.ns.$purchases VALUES (2, 10.0, cast('2021-01-01' as timestamp))")
withSQLConf(
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
- SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true",
+ // The explained key count is the union of both sides' reduced keys.
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "false") {
Review Comment:
**Finding 9.** This is the one pinned case left after finding 6, and I think it can be parameterized like the other seven, which finishes the ask at [issuecomment-5615352605](https://github.com/apache/spark/pull/58681#issuecomment-5615352605).
I flipped this line to `true` and ran the test. Only the count moves:
```
+- GroupPartitions JoinKeyPositions: [0] ExpectedPartitionKeys: 0 Reducers: [BucketReducer(2)] DistributePartitions: false SortedMerge: false
" did not contain "... ExpectedPartitionKeys: 2 Reducers: [BucketReducer(2)] ..."
```
`items` holds `id=1`, so `bucket(4, 1) = 1` reduces to `1 % 2 = 1`; `purchases` holds `item_id=2`, so `bucket(6, 2) = 2` reduces to `2 % 2 = 0`. The two reduced key sets are disjoint, so the inner join's union is `[0, 1]` and its intersection is empty. Nothing is wrong here: the join can produce no row either way, and the test asserts only the explain string.
That is also why your comment's reason does not quite carry the pin. If the count is incidental to what this test is about, which is that the node renders all five fields in all three explain modes, then asserting it at both values costs nothing. And pinned to `false` the test never renders the node under the shipped default at all, which is the property the other seven were parameterized for.
```scala
Seq(2 -> false, 0 -> true).foreach { case (expectedKeys, filter) =>
withSQLConf(
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true",
// Both sides' reduced keys are disjoint, so filtering leaves the join no key group at all.
SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> filter.toString) {
```
with `ExpectedPartitionKeys: $expectedKeys` in the two keyword strings. Worth saying what this does not buy: the empty-intersection *behaviour* is already covered by `SPARK-48949: test partition filters with no matches`, which sets the config explicitly. What is uncovered is the rendering, and the rendering at the default.
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.sorting.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, Spark satisfies a sort on the partition key expressions from the partitioning reported by a V2 data source, so no shuffle is added for that sort. The parallelism of the sorted output is then whatever the data source's partition layout provides, and there is no shuffle stage left for adaptive partition coalescing or skew splitting to balance. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
Review Comment:
**Finding 8.** `GroupPartitionsExec.outputOrdering` gates the key-expression filter on `reducers.isEmpty` as well as on this config (`sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:527`). With reducers the merged partitions share only the reduced key, so no original key expression is constant in them and the node reports nothing at all. A user running with `spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled` on gets no ordering out of this config on the joins that reduced, and the row gives them no way to know.
The migration-guide entry has the same gap (`docs/sql-migration-guide.md:34`), and its "since those expressions are constant within the merged partition" is exactly the clause that stops holding there.
```suggestion
When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. No order is reported when the join reduced the partition keys onto a common key space (see <code>spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled</code>), because the merged partitions then share only the reduced key. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
```
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.sorting.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, Spark satisfies a sort on the partition key expressions from the partitioning reported by a V2 data source, so no shuffle is added for that sort. The parallelism of the sorted output is then whatever the data source's partition layout provides, and there is no shuffle stage left for adaptive partition coalescing or skew splitting to balance. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> may preserve the child's full ordering through a sorted merge instead of concatenation, rather than only the orderings over partition key expressions that <code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code> preserves. The sorted merge is selected only where a downstream ordering is otherwise unsatisfied and the merge is feasible, that is, the node coalesces partitions sharing a key, the child reports a non-empty ordering, and its subtree is <code>SafeForKWayMerge</code>; otherwise the node concatenates. Where it applies, it removes a downstream sort when data is both partitioned and sorted, but a sorted merge costs more than concatenation, especially when merging many partitions, and it gives up columnar execution for the merged plan. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
Review Comment:
**Finding 7.** `SafeForKWayMerge` is a Scala marker trait in `sql/core`. It shows up in no plan string, no config and no other doc, so a reader cannot tell whether their subtree has it. The condition it stands for is describable: `childIsSafeForKWayMerge` (`sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:396`) asks that every operator below the node is one the merge can drive, because the merge holds all its input iterators open at once and an operator keeping per-partition state in a field would share it across them.
The qualification itself is right and it answers [r3995045894](https://github.com/apache/spark/pull/58681#discussion_r3995045894) — only the naming is worth changing.
```suggestion
When enabled, <code>GroupPartitionsExec</code> may preserve the child's full ordering through a sorted merge instead of concatenation, rather than only the orderings over partition key expressions that <code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code> preserves. The sorted merge is selected only where a downstream ordering is otherwise unsatisfied and the merge is feasible, that is, the node coalesces partitions sharing a key, the child reports a non-empty ordering, and every operator below it is one Spark can drive from several partitions at once; otherwise the node concatenates. Where it applies, it removes a downstream sort when data is both partitioned and sorted, but a sorted merge costs more than concatenation, especially when merging many partitions, and it gives up columnar execution for the merged plan. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58681:
URL: https://github.com/apache/spark/pull/58681#issuecomment-5754525530
Thanks @peter-toth and @cloud-fan for the re-checks and approvals.
- Finding 10: b0df3a4198b adds the sort-aggregate clause to the `partitionKeyOrdering` migration entry, worded as suggested.
- Finding 11: the description's testing section is updated: `SPARK-55992` is an eighth parameterized case, the `SPARK-59050` pin went away with the rebase (#58858 had already parameterized it), and the only explicit `false` left is the `SPARK-59256` pairing test's rollback arm.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you closed pull request #58681: [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default
URL: https://github.com/apache/spark/pull/58681
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58681:
URL: https://github.com/apache/spark/pull/58681#issuecomment-5615352605
Thank you for the PR, @ulysses-you. The direction makes sense to me. A few comments:
1. The PR description says "Yes, within unreleased branches", but `partition.filter.enabled` has defaulted to `false` since 4.0.0 and the two ordering configs since 4.2.0 (released). This is a user-facing behavior change relative to released versions. Could you update the description? (The migration guide entries are correct.)
2. Several existing tests are now pinned to `partition.filter.enabled=false` (SPARK-42038 DPP, SPARK-57881 union, SPARK-55535, SPARK-59050, ...). Since they used to cover the default path, could we parameterize them over `Seq(true, false)` with the expected partition counts for `true`, at least for the DPP and union cases? That would also verify the claim that the new default never adds shuffles.
3. nit: In the migration guide, "a join that preserves the left side's rows" is not precise for LeftAnti/ExistenceJoin. Maybe "a join that keeps or tests every left row", and mention that an equi CROSS join is treated like INNER.
cc @peter-toth @szehon-ho @pan3793 since the filtering for more join types (SPARK-59199) and the ordering derivation (SPARK-56241) are recent.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r4055790255
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -8191,17 +8234,27 @@ class KeyGroupedPartitioningSuite
withSQLConf(
SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
"spark.sql.autoBroadcastJoinThreshold" -> "-1",
- SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ // Filtering narrows the joined groups to the intersection, which leaves 2 groups over
+ // the marked side's 3 input partitions. The grouping stays index-identity but forfeits
+ // its keyed claim (SPARK-59272), and this test pins that claim surviving a regrouping,
+ // so it runs with filtering off.
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "false") {
Review Comment:
Rebased onto master (d39cc1784c0) and resolved the hunk with master's side, as suggested: the pin and the count-clause tightening are gone, master's `Seq(false, true)` loop with `numPartitions === 3` carries the assertion. One extra conflict beyond your two: #58659 renamed the node's `expectedPartitionKeys` field to `expectedKeyCount` after your recheck, resolved by combining the rename with the pairing test's loop.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5235,7 +5259,9 @@ class KeyGroupedPartitioningSuite
sql(s"INSERT INTO testcat.ns.$purchases VALUES (2, 10.0, cast('2021-01-01' as timestamp))")
withSQLConf(
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
- SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true",
+ // The explained key count is the union of both sides' reduced keys.
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "false") {
Review Comment:
Parameterized in a3ad406dadc as sketched: `Seq(2 -> false, 0 -> true)`, `ExpectedPartitionKeys: $expectedKeys` in both keyword strings, with a comment naming the disjoint reduced keys. That was the last pinned case.
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.sorting.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, Spark satisfies a sort on the partition key expressions from the partitioning reported by a V2 data source, so no shuffle is added for that sort. The parallelism of the sorted output is then whatever the data source's partition layout provides, and there is no shuffle stage left for adaptive partition coalescing or skew splitting to balance. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> may preserve the child's full ordering through a sorted merge instead of concatenation, rather than only the orderings over partition key expressions that <code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code> preserves. The sorted merge is selected only where a downstream ordering is otherwise unsatisfied and the merge is feasible, that is, the node coalesces partitions sharing a key, the child reports a non-empty ordering, and its subtree is <code>SafeForKWayMerge</code>; otherwise the node concatenates. Where it applies, it removes a downstream sort when data is both partitioned and sorted, but a sorted merge costs more than concatenation, especially when merging many partitions, and it gives up columnar execution for the merged plan. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
Review Comment:
Took the suggested wording in 3533901e52a: the row now states the feasibility condition ("every operator below it is one Spark can drive from several partitions at once") instead of naming the trait.
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.sorting.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, Spark satisfies a sort on the partition key expressions from the partitioning reported by a V2 data source, so no shuffle is added for that sort. The parallelism of the sorted output is then whatever the data source's partition layout provides, and there is no shuffle stage left for adaptive partition coalescing or skew splitting to balance. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
Review Comment:
Took the suggested row wording in 3533901e52a, and qualified the migration-guide entry the same way: a join that reduced the partition keys onto a common key space reports no order, since the merged partitions then share only the reduced key.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1779,19 +1779,26 @@ class EnsureRequirementsSuite extends SharedSparkSession {
val right = new DummySparkPlanWithBatchScanChild(
outputPartitioning = KeyedPartitioning(Seq(days(aR), years(bR)), rightKeys))
- // No `withSQLConf` on purpose. `pushPartValues` is on by default, which is all the push branch
- // needs, so this is what a user gets out of the box.
- val smj = SortMergeJoinExec(Seq(xL, yL), Seq(aR, bR), Inner, None, left, right)
- val planned = EnsureRequirements.apply(smj)
-
- // Without the pairing, the left's first member is taken and its `days(yL)` is matched against
- // `days(aR)`, which names the other join key, so the join declines and both sides are shuffled
- // onto the default partitioning.
- assert(planned.collect { case s: ShuffleExchangeExec => s }.isEmpty,
- "the second member pairs with the other side, so neither side is shuffled")
- assert(groupPartitionsNodes(planned).map(_.expectedPartitionKeys.map(_.size)) ===
- Seq(Some(3), Some(3)),
- "both sides are pushed the union of the two key sets")
+ // `pushPartValues` stays at its default (on), which is all the push branch needs. The first
+ // arm leaves `partition.filter` unset, so it exercises the shipped default and would fail if
+ // `createWithDefault(true)` were reverted; the second arm pins the off rollback.
+ val filterKey = SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key
+ Seq(1 -> Option.empty[String], 3 -> Some("false")).foreach { case (expected, filterOverride) =>
+ withSQLConf(filterOverride.map(filterKey -> _).toSeq: _*) {
Review Comment:
The unset arm is deliberate: it pins the shipped default, so it fails if `createWithDefault(true)` is reverted, which an explicit "true" arm would not. Rewritten in a3ad406dadc without the Option loop: two direct calls to a local helper, one with no conf (unset default, intersection 1) and one pinned to false (union 3).
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
Review Comment:
Took the suggested wording in 3533901e52a, and qualified the migration-guide entry to match: groups "may be narrowed", plus a sentence naming the skip -- filtering does not run when either side's partitioning may contain unknown partition keys, as after a shuffle on one side.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58681:
URL: https://github.com/apache/spark/pull/58681#issuecomment-5612511220
cc @peter-toth @dongjoon-hyun @cloud-fan thank you
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r4057205314
##########
docs/sql-migration-guide.md:
##########
@@ -34,6 +34,9 @@ license: |
- Since Spark 4.4, the CSV `extension` option must be non-empty and contain only letters. In Spark 4.3 and earlier, due to a bug in the extension validation check, empty strings and three-character suffixes containing non-letters (for example `ab1`) were accepted. This affects both writes and reads of CSV tables whose persisted `OPTIONS` contain such values; update the option to a non-empty letters-only suffix to restore access.
- Since Spark 4.4, when `array_repeat` or `array_insert` is asked to build an array larger than the maximum supported array length, generated code raises the same error as interpreted evaluation. `array_repeat` now fails with `COLLECTION_SIZE_LIMIT_EXCEEDED.PARAMETER` instead of the internal error `_LEGACY_ERROR_TEMP_2176`, and `array_insert` fails with `COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION` instead of `COLLECTION_SIZE_LIMIT_EXCEEDED.PARAMETER`, which named a `count` parameter that `array_insert` does not have. Both functions raise an error under exactly the same conditions as before; only the reported error condition changes.
- Since Spark 4.4, `/*+ ... */` inside a bracketed comment is parsed as a nested comment, so its closing `*/` no longer closes the outer comment. For example, `/* note /*+ x */ SELECT 1` previously returned `1`, but now raises `UNCLOSED_BRACKETED_COMMENT`. Close the outer comment explicitly, for example `/* note /*+ x */ */ SELECT 1`.
+- Since Spark 4.4, `spark.sql.sources.v2.bucketing.partition.filter.enabled` defaults to `true`. In a storage-partitioned join, the partition key groups pushed down to both sides may now be narrowed to those that can produce output for the join type, instead of always taking the union of the two sides' groups: an inner or semi join may keep only the groups present on both sides, a join that keeps or tests every left row (left outer, left anti, left single, existence) keeps the left side's groups, a right outer join keeps the right side's, and a full outer join is unaffected; a CROSS join carrying an equality condition is treated like an inner join. The narrowing is skipped when either side's partitioning may contain unknown partition keys, as after a shuffle on one side, so such a join keeps the full union. A group that is dropped is not scanned at all, so a query may read fewer files. To restore the previous behavior, set `spark.sql.sources.v2.bucketing.partition.filter.enabled` to
`false`.
+- Since Spark 4.4, `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` defaults to `true`. A V2 scan that reports a keyed partitioning but no explicit ordering now also reports itself sorted by its partition key expressions, because every row of such a partition evaluates them to the same value. Spark can then drop a `Sort` it would otherwise place above the scan. To restore the previous behavior, set `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` to `false`.
Review Comment:
**Finding 10.** A user whose plan changed under these defaults will most often be looking at an aggregate, not at a `Sort`. All three entries and the description's user-facing section name the visible change as a `Sort` disappearing. There is a second one: `spark.sql.execution.replaceHashWithSortAgg` is on by default and keys off `child.outputOrdering`, so once the scan reports its key expressions an aggregate whose grouping is exactly those expressions is planned as a sort aggregate.
Measured on `a3ad406`. `items` partitioned by `identity(id)` holding ids `1, 1, 2, 3`, `SELECT id, count(*) FROM items GROUP BY id`, AQE off, only these two configs varying. The rows are identical in all four arms:
| `partitionKeyOrdering` | `preserveKeyOrderingOnCoalesce` | aggregates |
|---|---|---|
| off | off | 2 hash |
| **on** | off | 1 hash, 1 sort |
| off | on | 2 hash |
| **on** | **on** | 2 sort |
The partial aggregate reads the scan, so this entry alone moves that one. The final aggregate reads the `GroupPartitionsExec` that coalesces the two `id = 1` splits, so the next entry moves that one. `SELECT DISTINCT id` behaves the same way. No sort is added in any arm, and with one key value per partition the sort aggregate is the cheaper shape, so this is not a regression. It is simply the plan diff a user is most likely to open the guide to explain.
One clause on this entry would cover it, for example after "Spark can then drop a `Sort` it would otherwise place above the scan.":
> An aggregate whose grouping is exactly the partition key expressions may also be planned as a sort aggregate rather than a hash aggregate, because `spark.sql.execution.replaceHashWithSortAgg` keys off the reported ordering.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58681:
URL: https://github.com/apache/spark/pull/58681#issuecomment-5616547282
Thanks Dongjoon.
1. Description fixed: `partition.filter.enabled` shipped opt-in in 4.0.0 and the two ordering
configs in 4.2.0, all released, so this is a behavior change relative to released Spark. The
"within unreleased branches" wording is gone.
2. 8dcbe648cf6 parameterizes seven of the pinned cases over
`spark.sql.sources.v2.bucketing.partition.filter.enabled`, pinning the aligned key count for
each setting and asserting that neither setting adds a shuffle: SPARK-42038 with dynamic
partition filtering (15/6 to 9/3), both SPARK-57881 union cases (5 to 3, 4 to 2), SPARK-55535
granular grouping (Seq(3, 6, 6, 3) to Seq(1, 2, 2, 1)) and partial clustering
(Seq(6, 5, 5, 6) to Seq(2, 2, 2, 2)), SPARK-48065 (8/4 to 6/2), and SPARK-56182 (4 to 3).
Two stay pinned: SPARK-55992 asserts the explain string of a padded leg, and SPARK-59050 pins
the unknown-keyed claim, which is finding 1 in @peter-toth's review.
3. Reworded to "a join that keeps or tests every left row (left outer, left anti, left single,
existence)", and added that a CROSS join carrying an equality condition behaves like an inner
join.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r3977486130
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -8185,7 +8204,10 @@ class KeyGroupedPartitioningSuite
withSQLConf(
SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
"spark.sql.autoBroadcastJoinThreshold" -> "-1",
- SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ // The identity grouping only appears when the marked side is padded up to the union
+ // of both sides' keys instead of being narrowed to their intersection.
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "false") {
Review Comment:
**Finding 1.** The comment misreads why this test fails with filtering on, and the pin is what hides finding 2.
I ran the test with only this line changed to `"true"`. `identityGpe` is still found, so the identity grouping does still appear. The failure is one assertion later, at line 8229:
```
- SPARK-59050: SPJ: a reducer-free identity regrouping keeps the unknown-keyed claim *** FAILED ***
expected a KeyedPartitioning output, got UnknownPartitioning(2) (KeyGroupedPartitioningSuite.scala:8229)
```
The test's own predicate only checks that output partition i holds input partition i, which the intersection satisfies. `identityGrouping` asks more: it also requires `grouping.partitions.size == child.outputPartitioning.numPartitions`, and the intersection leaves 2 groups over the marked side's 3 input partitions. So the marked layout gives up its keyed claim, which is finding 2, on the very test that exists to pin that claim surviving a regrouping.
Two things follow. The comment should say that, not that the identity grouping disappears. And this test should keep running on the shipping default, because the branch it pins is reachable there. If finding 2 is closed by not filtering a marked side, the pin can go away entirely.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2547,7 +2547,7 @@ object SQLConf {
s"enabled.")
.version("4.0.0")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
Review Comment:
**Finding 2.** This flip makes SPARK-59272 reachable one opt-in away from the defaults, instead of two.
Partition filtering is the only path that shrinks the pushed-down key list *below* a side's own declared keys. `pushPartValues` padding always merges to a superset, so no side ever has to drop one of its groups. An intersection can, and then `alignToExpectedKeys` leaves that side with fewer groups than it has input partitions. `identityGrouping` rejects that on the count clause, so a marked `KeyedPartitioning` becomes `UnknownPartitioning` (`GroupPartitionsExec.scala:101`). Its own doc names this exact case: "A grouping that drops trailing declared keys still reads identity for every group it keeps, but the partition count shrinks and the hash modulus with it."
The give-up is meant to make the parent shuffle. Nothing reads it here, which is the gap you described at [r3939155993](https://github.com/apache/spark/pull/58339#discussion_r3939155993) and filed as SPARK-59272: "the committed pair then carries a child `ValidateRequirements` rejects, at the cost of AQE rules on that stage". That ticket is still open and unassigned.
Measured in a worktree at `c2c91ea`, on the query from your own test at `KeyGroupedPartitioningSuite.scala:8177` (`pa`=[1,2,3], `pt`=[1,2,3,4], `pb`=[1,2], `v2.bucketing.shuffle.enabled=true`, AQE off), reading `ValidateRequirements.validate(df.queryExecution.executedPlan)`:
| `partition.filter.enabled` | left `GroupPartitionsExec.outputPartitioning` | `validate` |
|---|---|---|
| `false` (today's default) | `KeyedPartitioning`, 3 groups over 3 input partitions | `true` |
| `true` (this PR) | `UnknownPartitioning(2)`, 2 groups over 3 | **`false`** |
Both arms keep exactly one shuffle, so the join above the give-up node is never re-planned.
**Why this is not Blocking, so you can weigh it as I do.** Results stay correct, and not by luck. `areKeysCompatible`'s marked path forces the unmarked side's declared keys to be a subset of the marked side's, so the merged list can never hold a key the marked side does not declare, so the marked side's undeclared rows can never match anything on the other side wherever the regrouping puts them. I tried the shape that would break that (`pb`=[1,2,4] against a marked [1,2,3]) and the gate refuses the pairing there, both sides shuffle. The cost is the one you named: `ValidateRequirements` is read only by `AdaptiveSparkPlanExec:209` and `OptimizeSkewedJoin:261`, and both revert their change when it fails, so coalescing, local read, rebalance skew and skew join are all silently skipped for a stage holding a give-up node. I could not exhibit that loss. My attempt put a coalescable hash exchange in the same stage, and the control arm with no SPJ at all produced no coalesced read either,
so the harness measured nothing.
So what I can show is the invariant violation and the ticket, not a wrong answer or a measured regression. Worth deciding on rather than merging past, but I am not blocking on it.
**#58659 implements the pairwise check SPARK-59272 asks for**, which I should disclose is my own draft PR. It compares the two rebuilt children's declared key sequences before committing the pairing, so it is the joint-alignment check you described rather than a per-side `satisfies`. Same probe on its head `27ef3a60f1f`:
| `partition.filter.enabled` | plan | `validate` |
|---|---|---|
| `false` | 1 shuffle, 2 `GroupPartitionsExec` | `true` |
| `true` | 2 shuffles, 0 `GroupPartitionsExec` | `true` |
It is a draft though, and it is stacked on #58531 and #58552, so it is not a short wait.
Three ways out, in the order I would pick them:
1. Land the two ordering flips now and hold `partition.filter.enabled` until SPARK-59272 is closed. Nothing I found argues against the other two.
2. Keep all three and stop filtering a marked side, so this PR does not depend on my stack. This is not the per-side `satisfies` you ruled out, it is a guard on the merge input, so partial clustering is untouched. It is also the principled version: the intersection argument assumes a side's declared key list accounts for all of its rows, and `mayContainUnknownPartitionKeys` says it does not.
```scala
// EnsureRequirements.scala, at the only call site (~:620)
var mergedPartitionKeys =
mergeAndDedupPartitions(leftReducedKeys, rightReducedKeys, joinType, reducedKeyOrdering,
// A marked side holds rows whose key is not in its declared list, so an intersection cannot
// reason about them, and dropping one of its groups forfeits the keyed claim it needs.
filter = !leftPartitioning.mayContainUnknownPartitionKeys &&
!rightPartitioning.mayContainUnknownPartitionKeys)
.map((_, 1))
```
A new trailing parameter with a default keeps the `EnsureRequirementsSuite` call site as it is.
3. Keep all three and say in the description that the flip puts SPARK-59272 one opt-in from the defaults. That drops the "None of them adds runtime cost" claim for this one.
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires both <code>spark.sql.sources.v2.bucketing.enabled</code> and <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.sorting.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, Spark satisfies a sort on the partition key expressions from the partitioning reported by a V2 data source, so no shuffle is added for that sort. The parallelism of the sorted output is then whatever the data source's partition layout provides, and there is no shuffle stage left for adaptive partition coalescing or skew splitting to balance. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> merges partitions that share a key by a sorted merge rather than by concatenation, so it can report the child's full ordering instead of only the orderings over partition key expressions that <code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code> preserves. This removes a downstream sort when data is both partitioned and sorted, but a sorted merge costs more than concatenation, especially when merging many partitions. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
Review Comment:
**Finding 3.** This row gives the per-row cost of the sorted merge but not the larger one: enabling it gives up columnar execution. `GroupPartitionsExec.supportsColumnar` is `child.supportsColumnar && !usesSortedMerge` (`GroupPartitionsExec.scala:488`), so a vectorized Parquet or Iceberg scan under the merge gains a `ColumnarToRow`. Your PR description states this cost, the row a user reads does not.
```suggestion
When enabled, <code>GroupPartitionsExec</code> merges partitions that share a key by a sorted merge rather than by concatenation, so it can report the child's full ordering instead of only the orderings over partition key expressions that <code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code> preserves. This removes a downstream sort when data is both partitioned and sorted, but a sorted merge costs more than concatenation, especially when merging many partitions, and it gives up columnar execution for the merged plan. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
```
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires both <code>spark.sql.sources.v2.bucketing.enabled</code> and <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> to be true.
Review Comment:
**Finding 4.** `pushPartValues.enabled` is not strictly required. `EnsureRequirements.scala:548` enters the merge block on `v2BucketingPushPartValuesEnabled || v2BucketingAllowKeysSubsetOfPartitionKeys`, and `mergeAndDedupPartitions` is the only reader of this config. So with `pushPartValues.enabled` false and `allowKeysSubsetOfPartitionKeys.enabled` true, filtering still applies.
The same wording is in the `.doc` string, so it is pre-existing, but this PR is the one writing it into the user-facing guide.
```suggestion
When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58681:
URL: https://github.com/apache/spark/pull/58681#issuecomment-5657947590
Thanks @cloud-fan. The three findings are answered on their inline threads. The description is
also updated: partition filtering requires `spark.sql.sources.v2.bucketing.enabled` plus either
`pushPartValues.enabled` or `allowKeysSubsetOfPartitionKeys.enabled`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r4001691044
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2542,12 +2542,13 @@ object SQLConf {
buildConf("spark.sql.sources.v2.bucketing.partition.filter.enabled")
.doc(s"Whether to filter partitions when running storage-partition join. " +
s"When enabled, partitions without matches on the other side can be omitted for " +
- s"scanning, if allowed by the join type. This config requires both " +
- s"${V2_BUCKETING_ENABLED.key} and ${V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key} to be " +
- s"enabled.")
+ s"scanning, if allowed by the join type. This config requires " +
+ s"${V2_BUCKETING_ENABLED.key} to be enabled, together with either " +
+ s"${V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key} or " +
+ s"${V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key}.")
.version("4.0.0")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
val V2_BUCKETING_SORTING_ENABLED =
buildConf("spark.sql.sources.v2.bucketing.sorting.enabled")
Review Comment:
Keeping the agreed landing order: this PR will not merge until SPARK-59272 / #58659 lands, with no
guard added here. This is also the condition of @sunchao's approval. Thanks both.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1779,19 +1779,24 @@ class EnsureRequirementsSuite extends SharedSparkSession {
val right = new DummySparkPlanWithBatchScanChild(
outputPartitioning = KeyedPartitioning(Seq(days(aR), years(bR)), rightKeys))
- // No `withSQLConf` on purpose. `pushPartValues` is on by default, which is all the push branch
- // needs, so this is what a user gets out of the box.
- val smj = SortMergeJoinExec(Seq(xL, yL), Seq(aR, bR), Inner, None, left, right)
- val planned = EnsureRequirements.apply(smj)
-
- // Without the pairing, the left's first member is taken and its `days(yL)` is matched against
- // `days(aR)`, which names the other join key, so the join declines and both sides are shuffled
- // onto the default partitioning.
- assert(planned.collect { case s: ShuffleExchangeExec => s }.isEmpty,
- "the second member pairs with the other side, so neither side is shuffled")
- assert(groupPartitionsNodes(planned).map(_.expectedPartitionKeys.map(_.size)) ===
- Seq(Some(3), Some(3)),
- "both sides are pushed the union of the two key sets")
+ // `pushPartValues` is left at its default (on), which is all the push branch needs, so this is
+ // what a user gets out of the box. `partition.filter` is varied to cover both settings: off
+ // pushes the union of the two key sets, on (the default) their intersection.
+ Seq(3 -> false, 1 -> true).foreach { case (expected, filter) =>
+ withSQLConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> filter.toString) {
Review Comment:
Fixed in 8c5c30e0807: the enabled arm now leaves `partition.filter.enabled` unset and asserts the
intersection (`Some(1)`), so reverting `createWithDefault(true)` fails it; the explicit `false`
arm (union, `Some(3)`) stays as the rollback.
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.sorting.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, Spark satisfies a sort on the partition key expressions from the partitioning reported by a V2 data source, so no shuffle is added for that sort. The parallelism of the sorted output is then whatever the data source's partition layout provides, and there is no shuffle stage left for adaptive partition coalescing or skew splitting to balance. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> merges partitions that share a key by a sorted merge rather than by concatenation, so it can report the child's full ordering instead of only the orderings over partition key expressions that <code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code> preserves. This removes a downstream sort when data is both partitioned and sorted, but a sorted merge costs more than concatenation, especially when merging many partitions, and it gives up columnar execution for the merged plan. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
Review Comment:
Fixed in 8c5c30e0807: the row now says `GroupPartitionsExec` *may* select the sorted merge, gated
on a downstream ordering otherwise unsatisfied plus a feasible merge (coalescing, non-empty child
ordering, `SafeForKWayMerge` subtree); otherwise it concatenates.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r3977724625
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -8185,7 +8204,10 @@ class KeyGroupedPartitioningSuite
withSQLConf(
SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
"spark.sql.autoBroadcastJoinThreshold" -> "-1",
- SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ // The identity grouping only appears when the marked side is padded up to the union
+ // of both sides' keys instead of being narrowed to their intersection.
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "false") {
Review Comment:
You are right, my comment had the mechanism backwards. 8dcbe648cf6 rewrites it: with filtering the
grouping is still index-identity, and it is the intersection leaving 2 groups over the marked
side's 3 input partitions that trips the give-up at `GroupPartitionsExec.scala:101`. The test
predicate now also requires `groupedPartitions.size == child.outputPartitioning.numPartitions`,
the second thing `identityGrouping` asks, so the predicate no longer over-claims. The pin stays
for now, for the reason in my reply to Finding 2.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2547,7 +2547,7 @@ object SQLConf {
s"enabled.")
.version("4.0.0")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
Review Comment:
Agreed on the whole read, including that correctness rests on `areKeysCompatible`'s subset rule
rather than on the give-up, and that the cost is AQE rules silently skipping the stage.
Decision: this PR keeps all three flips and adds no guard. SPARK-59396 targets 4.4.0, so it can
land after SPARK-59272 and your #58659 line.
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires both <code>spark.sql.sources.v2.bucketing.enabled</code> and <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.sorting.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, Spark satisfies a sort on the partition key expressions from the partitioning reported by a V2 data source, so no shuffle is added for that sort. The parallelism of the sorted output is then whatever the data source's partition layout provides, and there is no shuffle stage left for adaptive partition coalescing or skew splitting to balance. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> merges partitions that share a key by a sorted merge rather than by concatenation, so it can report the child's full ordering instead of only the orderings over partition key expressions that <code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code> preserves. This removes a downstream sort when data is both partitioned and sorted, but a sorted merge costs more than concatenation, especially when merging many partitions. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
Review Comment:
Adopted your wording verbatim in 8dcbe648cf6.
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires both <code>spark.sql.sources.v2.bucketing.enabled</code> and <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> to be true.
Review Comment:
Adopted, and fixed at the source too: the same "requires both" wording in the
`V2_BUCKETING_PARTITION_FILTER_ENABLED` doc string now reads `enabled` together with either
`spark.sql.sources.v2.bucketing.pushPartValues.enabled` or
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r4057252437
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2542,12 +2542,13 @@ object SQLConf {
buildConf("spark.sql.sources.v2.bucketing.partition.filter.enabled")
.doc(s"Whether to filter partitions when running storage-partition join. " +
s"When enabled, partitions without matches on the other side can be omitted for " +
- s"scanning, if allowed by the join type. This config requires both " +
- s"${V2_BUCKETING_ENABLED.key} and ${V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key} to be " +
- s"enabled.")
+ s"scanning, if allowed by the join type. This config requires " +
+ s"${V2_BUCKETING_ENABLED.key} to be enabled, together with either " +
+ s"${V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key} or " +
+ s"${V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key}.")
.version("4.0.0")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
val V2_BUCKETING_SORTING_ENABLED =
buildConf("spark.sql.sources.v2.bucketing.sorting.enabled")
Review Comment:
Resolved at the pinned head: SPARK-59272/#58659 is now in the base, and the pair-selection/unknown-key guards prevent committing a filtered child that loses its required distribution.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:3995045887","thread_id":"inline:3995045887","verdict_sha256":"27e2ae0e4489daef7ab15c492a4719b0eb3a2c1d325bf16fb32d1f2054b034b9"} -->
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1779,19 +1779,24 @@ class EnsureRequirementsSuite extends SharedSparkSession {
val right = new DummySparkPlanWithBatchScanChild(
outputPartitioning = KeyedPartitioning(Seq(days(aR), years(bR)), rightKeys))
- // No `withSQLConf` on purpose. `pushPartValues` is on by default, which is all the push branch
- // needs, so this is what a user gets out of the box.
- val smj = SortMergeJoinExec(Seq(xL, yL), Seq(aR, bR), Inner, None, left, right)
- val planned = EnsureRequirements.apply(smj)
-
- // Without the pairing, the left's first member is taken and its `days(yL)` is matched against
- // `days(aR)`, which names the other join key, so the join declines and both sides are shuffled
- // onto the default partitioning.
- assert(planned.collect { case s: ShuffleExchangeExec => s }.isEmpty,
- "the second member pairs with the other side, so neither side is shuffled")
- assert(groupPartitionsNodes(planned).map(_.expectedPartitionKeys.map(_.size)) ===
- Seq(Some(3), Some(3)),
- "both sides are pushed the union of the two key sets")
+ // `pushPartValues` is left at its default (on), which is all the push branch needs, so this is
+ // what a user gets out of the box. `partition.filter` is varied to cover both settings: off
+ // pushes the union of the two key sets, on (the default) their intersection.
+ Seq(3 -> false, 1 -> true).foreach { case (expected, filter) =>
+ withSQLConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> filter.toString) {
Review Comment:
Resolved: the pairing test now leaves the enabled arm unset, so reverting `createWithDefault(true)` fails, and retains an explicit `false` rollback arm.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:3995045892","thread_id":"inline:3995045892","verdict_sha256":"27e2ae0e4489daef7ab15c492a4719b0eb3a2c1d325bf16fb32d1f2054b034b9"} -->
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.sorting.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, Spark satisfies a sort on the partition key expressions from the partitioning reported by a V2 data source, so no shuffle is added for that sort. The parallelism of the sorted output is then whatever the data source's partition layout provides, and there is no shuffle stage left for adaptive partition coalescing or skew splitting to balance. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.0.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, Spark derives the output ordering of a V2 scan from its partition key expressions, if the source reports a keyed partitioning but no explicit ordering. All rows of such a partition share one key value, so the partition is trivially sorted by those expressions, and a sort Spark would otherwise add becomes unnecessary. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> reports sort orders over partition key expressions after coalescing several input partitions into one. The merged partitions share the same partition key value, so these orders still hold, while orders over other columns are lost by the concatenation. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
+ </td>
+ <td>4.2.0</td>
+ </tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled</code></td>
+ <td>false</td>
+ <td>
+ When enabled, <code>GroupPartitionsExec</code> merges partitions that share a key by a sorted merge rather than by concatenation, so it can report the child's full ordering instead of only the orderings over partition key expressions that <code>spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled</code> preserves. This removes a downstream sort when data is both partitioned and sorted, but a sorted merge costs more than concatenation, especially when merging many partitions, and it gives up columnar execution for the merged plan. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true.
Review Comment:
Resolved: the row now makes sorted merge conditional on downstream demand and feasibility, and says the node concatenates otherwise.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:3995045894","thread_id":"inline:3995045894","verdict_sha256":"27e2ae0e4489daef7ab15c492a4719b0eb3a2c1d325bf16fb32d1f2054b034b9"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r4042519572
##########
docs/sql-performance-tuning.md:
##########
@@ -688,6 +688,46 @@ The following SQL properties enable Storage Partition Join in different join que
</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>spark.sql.sources.v2.bucketing.partition.filter.enabled</code></td>
+ <td>true</td>
+ <td>
+ When enabled, key groups that cannot produce output for the join type are not scanned at all, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join only scans the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
Review Comment:
After rebasing over #58858, could we qualify "an inner join only scans the key groups present on both sides" here and in the partition-filtering migration-guide entry? `EnsureRequirements.filtersKeys` skips filtering when either side may contain unknown partition keys, as can happen after a shuffle on one side. Those joins retain the groups even with this setting enabled.
Suggested wording:
```suggestion
When enabled, key groups that cannot produce output for the join type may be skipped, instead of being filled with empty partitions on the side that does not hold them. For example, an inner join may scan only the key groups present on both sides. This config requires <code>spark.sql.sources.v2.bucketing.enabled</code> to be true, together with either <code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> or <code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code>.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1779,19 +1779,26 @@ class EnsureRequirementsSuite extends SharedSparkSession {
val right = new DummySparkPlanWithBatchScanChild(
outputPartitioning = KeyedPartitioning(Seq(days(aR), years(bR)), rightKeys))
- // No `withSQLConf` on purpose. `pushPartValues` is on by default, which is all the push branch
- // needs, so this is what a user gets out of the box.
- val smj = SortMergeJoinExec(Seq(xL, yL), Seq(aR, bR), Inner, None, left, right)
- val planned = EnsureRequirements.apply(smj)
-
- // Without the pairing, the left's first member is taken and its `days(yL)` is matched against
- // `days(aR)`, which names the other join key, so the join declines and both sides are shuffled
- // onto the default partitioning.
- assert(planned.collect { case s: ShuffleExchangeExec => s }.isEmpty,
- "the second member pairs with the other side, so neither side is shuffled")
- assert(groupPartitionsNodes(planned).map(_.expectedPartitionKeys.map(_.size)) ===
- Seq(Some(3), Some(3)),
- "both sides are pushed the union of the two key sets")
+ // `pushPartValues` stays at its default (on), which is all the push branch needs. The first
+ // arm leaves `partition.filter` unset, so it exercises the shipped default and would fail if
+ // `createWithDefault(true)` were reverted; the second arm pins the off rollback.
+ val filterKey = SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key
+ Seq(1 -> Option.empty[String], 3 -> Some("false")).foreach { case (expected, filterOverride) =>
+ withSQLConf(filterOverride.map(filterKey -> _).toSeq: _*) {
Review Comment:
why not just 'true' and 'false' like other ones, why do we need Option?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58681:
URL: https://github.com/apache/spark/pull/58681#issuecomment-5616547534
Thanks Peter. All four are addressed in 8dcbe648cf6, with a reply on each inline thread. Short
version: the SPARK-59050 comment and predicate now match the give-up mechanism (Finding 1); this
PR keeps all three flips and adds no guard, landing after SPARK-59272 and #58659 (Finding 2); and
both doc wordings are adopted (Findings 3 and 4).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58681:
URL: https://github.com/apache/spark/pull/58681#issuecomment-5759795308
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/7800e28711a00a5456b18805a249628fc623e1b1
- merged into branch-4.x https://github.com/apache/spark/commit/489d3db20724a76f188ced7d2a5a40dd27218db2
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r4058865311
##########
docs/sql-migration-guide.md:
##########
@@ -34,6 +34,9 @@ license: |
- Since Spark 4.4, the CSV `extension` option must be non-empty and contain only letters. In Spark 4.3 and earlier, due to a bug in the extension validation check, empty strings and three-character suffixes containing non-letters (for example `ab1`) were accepted. This affects both writes and reads of CSV tables whose persisted `OPTIONS` contain such values; update the option to a non-empty letters-only suffix to restore access.
- Since Spark 4.4, when `array_repeat` or `array_insert` is asked to build an array larger than the maximum supported array length, generated code raises the same error as interpreted evaluation. `array_repeat` now fails with `COLLECTION_SIZE_LIMIT_EXCEEDED.PARAMETER` instead of the internal error `_LEGACY_ERROR_TEMP_2176`, and `array_insert` fails with `COLLECTION_SIZE_LIMIT_EXCEEDED.FUNCTION` instead of `COLLECTION_SIZE_LIMIT_EXCEEDED.PARAMETER`, which named a `count` parameter that `array_insert` does not have. Both functions raise an error under exactly the same conditions as before; only the reported error condition changes.
- Since Spark 4.4, `/*+ ... */` inside a bracketed comment is parsed as a nested comment, so its closing `*/` no longer closes the outer comment. For example, `/* note /*+ x */ SELECT 1` previously returned `1`, but now raises `UNCLOSED_BRACKETED_COMMENT`. Close the outer comment explicitly, for example `/* note /*+ x */ */ SELECT 1`.
+- Since Spark 4.4, `spark.sql.sources.v2.bucketing.partition.filter.enabled` defaults to `true`. In a storage-partitioned join, the partition key groups pushed down to both sides may now be narrowed to those that can produce output for the join type, instead of always taking the union of the two sides' groups: an inner or semi join may keep only the groups present on both sides, a join that keeps or tests every left row (left outer, left anti, left single, existence) keeps the left side's groups, a right outer join keeps the right side's, and a full outer join is unaffected; a CROSS join carrying an equality condition is treated like an inner join. The narrowing is skipped when either side's partitioning may contain unknown partition keys, as after a shuffle on one side, so such a join keeps the full union. A group that is dropped is not scanned at all, so a query may read fewer files. To restore the previous behavior, set `spark.sql.sources.v2.bucketing.partition.filter.enabled` to
`false`.
+- Since Spark 4.4, `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` defaults to `true`. A V2 scan that reports a keyed partitioning but no explicit ordering now also reports itself sorted by its partition key expressions, because every row of such a partition evaluates them to the same value. Spark can then drop a `Sort` it would otherwise place above the scan. To restore the previous behavior, set `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` to `false`.
Review Comment:
Added in b0df3a4198b: the entry now says an aggregate whose grouping is exactly the partition key expressions may also be planned as a sort aggregate rather than a hash aggregate, because `spark.sql.execution.replaceHashWithSortAgg` keys off the reported ordering.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59396][SQL] Enable SPJ partition filtering and ordering over partition keys by default [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #58681:
URL: https://github.com/apache/spark/pull/58681#issuecomment-5747101741
Thank you @peter-toth, @szehon-ho, @cloud-fan, @dongjoon-hyun and @sunchao for the thorough reviews! All comments have been addressed and fixed.
The prerequisites have landed: SPARK-59272 (#58814, with @peter-toth's #58858) and SPARK-59289 (#58659), so the landing-order condition from the earlier reviews is satisfied, and the branch is rebased onto master (`d39cc1784c0`).
Two new commits answer the latest review round:
- `3533901e52a` (docs): the partition.filter row and migration entry now say key groups *may* be skipped and name the skip -- filtering does not run when either side's partitioning may contain unknown partition keys (`EnsureRequirements.filtersKeys`); the preserveKeyOrderingOnCoalesce row and entry now note that a join that reduced the partition keys reports no order (`outputOrdering` also gates on `reducers.isEmpty`); the sorted-merge row describes the feasibility condition directly instead of naming the internal `SafeForKWayMerge` trait.
- `a3ad406dadc` (tests): SPARK-55992 is parameterized over both filter settings (`ExpectedPartitionKeys` 2 vs 0), removing the last pinned case; the SPARK-59256 pairing test keeps covering the unset default and the pinned-off rollback as two direct calls to a local helper.
Note: the branch was force-pushed for the rebase; earlier replies referencing `8c5c30e0807` map to `e223f6a8ad1` after it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org