cloud-fan commented on code in PR #58681:
URL: https://github.com/apache/spark/pull/58681#discussion_r3995045887
##########
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:
**Non-blocking (P2):** This default flip makes the existing SPARK-59272 path
reachable with only `v2.bucketing.shuffle.enabled` opted in. If filtering
intersects a marked side below its declared partition count,
`GroupPartitionsExec` reports `UnknownPartitioning` after `EnsureRequirements`
has already committed the pair; `ValidateRequirements` then rejects AQE
shuffle-read and skew rewrites for that stage. Since the prerequisite fix is
still open, please keep the agreed landing order or guard this default-on path
so the selected children continue to satisfy the join distributions.
**Recommended change:** Make partition-filtered alignment accept a selected
pair only when each resulting child still satisfies its required distribution;
otherwise fall back to the existing safe, unfiltered alignment/shuffle path
instead of committing the marked GroupPartitionsExec plan.
**Why this works:** Evaluate distribution satisfaction after the filtering
and grouping transforms have produced their effective output partitioning. If a
marked side would take the GroupPartitionsExec give-up path and expose
UnknownPartitioning, discard that filtered candidate and continue through the
established non-filtered requirement-enforcement path. Keep filtering enabled
for candidates whose effective partitioning remains valid.
**Scope:** sql/core/src/main/scala/org/apache/spark/sql/execution/exchange,
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2,
sql/core/src/test/scala/org/apache/spark/sql/connector,
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2
**Compatibility:** Keep correct query results, valid non-marked partition
filtering, and the documented false opt-out unchanged.
**Risks:** The fallback can introduce a shuffle or retain more partitions
for the affected marked case, reducing the optimization benefit. A validation
placed before the effective GroupPartitionsExec output is known could repeat
the current false-positive selection.
**Constraints:** Do not change query answers or permit a child that fails
its join distribution requirement. Do not disable partition filtering for
unmarked or otherwise distribution-valid pairings. Preserve the explicit false
rollback behavior of the configuration.
**Success:** Every shuffle-free join pair selected after partition filtering
satisfies both required distributions. A marked side whose filtered grouping
would report UnknownPartitioning uses a valid fallback plan. Distribution-valid
partition-filtered joins retain their existing pruning and shuffle-avoidance
behavior.
##########
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:
**Non-blocking (P2):** This arm is described as covering the default, but
`withSQLConf` explicitly sets `V2_BUCKETING_PARTITION_FILTER_ENABLED`. The
other changed partition-filter cases also supply `true` or `false`, and
SPARK-59050 pins `false`, so reverting `createWithDefault(true)` would leave
the focused tests green. Could one distribution-valid planning case leave this
setting unset and assert the intended enabled plan/result, while keeping an
explicit `false` rollback 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> 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:
**Non-blocking (P2):** Enabling this setting does not by itself make
`GroupPartitionsExec` use sorted merge. `EnsureRequirements` only tries it when
a required ordering is otherwise unsatisfied, and the node additionally
requires coalescing, non-empty child ordering, and a `SafeForKWayMerge`
subtree; otherwise it still concatenates and does not report the full child
ordering. Could this say sorted merge *may* be selected when those demand and
feasibility conditions hold?
--
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]
