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: [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]
