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