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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
