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