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