You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "peter-toth (via GitHub)" <gi...@apache.org> on 2026/08/04 09:00:38 UTC
[PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
peter-toth opened a new pull request, #57753:
URL: https://github.com/apache/spark/pull/57753
### What changes were proposed in this pull request?
Follow-up to SPARK-40259 (subquery plan merge for DataSource V2 scans).
Today the DSv2 scan merge declines whenever either input scan reports key-grouped partitioning or ordering. The rebuilt merged scan carries neither on its own, so fusing the two scans could drop a partitioning/ordering the original plan relied on and force an extra shuffle or sort. Declining is safe but leaves the merge unused for any partitioned/ordered source.
This PR lets the merge proceed and re-derives the merged scan's own report instead of declining up front:
- Drop the `keyGroupedPartitioning`/`ordering` conjuncts from the `mergeable` gate in `PlanMerger.tryMergeScanRelations`.
- At the leaf, combine the two inputs' reports (np's remapped into cp's relation space) into the single report the merge must preserve:
- key-grouped partitioning: the two must be equal (`combineRequiredKeyGroupedPartitioning`);
- ordering: the stronger of the two, i.e. the one that satisfies the other via `SortOrder.orderingSatisfies` (`combineRequiredOrdering`).
If the inputs are incompatible (differing non-empty kGP, or neither ordering satisfies the other), no rebuilt scan can keep both not-worse, so the merge is declined right there -- before rebuilding -- saving a scan rebuild.
- After `V2ScanRelationPushDown.rebuildScan`, re-derive the merged scan's partitioning/ordering by running `V2ScanPartitioningAndOrdering` on the single merged scan node, then check it against the combined required report (`mergeDegradesReporting`): decline if the merged scan's kGP does not match, or its ordering does not satisfy, what was required. The required report is carried through `DSv2DeferredScan` for the deferred (under-`Filter`) build.
Two new opt-in configs gate accepting a degradation instead of declining (both default `false`), so with the defaults this is a pure improvement -- merge when not worse, decline on loss:
- `spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowKeyGroupedPartitioningDegradation`
- `spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowOrderingDegradation`
Only sources that declare the `SCAN_MERGING` table capability are affected.
### Why are the changes needed?
The parent feature (SPARK-40259) conservatively declines a merge whenever an input reports key-grouped partitioning or ordering, to avoid forcing a shuffle/sort that the original plan avoided. That is correct but overly broad: when the rebuilt merged scan re-derives the same (or a stronger) report, fusing the scans loses nothing and still removes a duplicate scan -- which matters for partitioned/ordered sources, e.g. an SPJ join inside the merged subplan. This PR keeps the safety (decline on real degradation) while capturing the merge when it is genuinely not worse.
### Does this PR introduce _any_ user-facing change?
No. The two new configs are opt-in and default to `false`, and no built-in source declares `SCAN_MERGING`, so there is no behavior change for existing sources.
### How was this patch tested?
New and existing unit tests, all under `sql/core`:
- `MergeSubplansSuite`:
- default-decline when a single input reports kGP/ordering the rebuilt scan drops (degradation);
- default-decline when the two inputs report incompatible kGP/ordering (declined at the leaf, before rebuild);
- config-allows-degradation: the merge proceeds to the plain column union under each `allow...Degradation=true`;
- the deferred (under-`Filter`) build path enforces the required report too: decline by default, merge with the config on;
- the pre-existing empty-report case still merges.
- `DSv2PlanMergingSuite`: an end-to-end test that a scan merge preserves the sources' reported key-grouped partitioning, using a new `SCAN_MERGING` fixture (`InMemoryScanMergingReportingCatalog`/`InMemoryScanMergingReportingTable`) that keeps its reported partitioning (no `NonReportingScan` wrapper), an identity-partitioned table under `spark.sql.sources.v2.bucketing.enabled=true`, and two scalar subqueries reading the partition column.
`build/sbt 'sql/testOnly *MergeSubplansSuite *DSv2PlanMergingSuite *PlanMergingSuite'` -- 95 tests pass. `dev/lint-scala` clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #57753:
URL: https://github.com/apache/spark/pull/57753#issuecomment-5230987713
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/8645e2c08929bbc1ef2ef66b442f525814f92220
- merged into branch-4.x https://github.com/apache/spark/commit/c060e1a26c330abef20d8393c0ce1aa4a8f1b79d
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3730362347
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2618,6 +2619,109 @@ class MergeSubplansSuite extends PlanTest {
assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
}
+ test("SPARK-58549: do not merge DSv2 scans reporting incompatible kGP/ordering") {
Review Comment:
@-
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7467,6 +7293,32 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val MERGE_SUBPLANS_DSV2_ALLOW_KEY_GROUPED_PARTITIONING_DEGRADATION = buildConf(
+ "spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowKeyGroupedPartitioningDegradation")
+ .doc("When false, a DataSource V2 scan merge is declined if the rebuilt merged scan would " +
Review Comment:
@-
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -760,6 +794,16 @@ class PlanMerger(
// re-checks it), and the scan must produce exactly the requested union of columns.
strictFilters.forall(ExpressionSet(scan.pushedFilters).contains) &&
scan.outputSet == AttributeSet(unionAttrs)
+ }.map { scan =>
+ // rebuildScan returns the merged scan with reported partitioning/ordering unset
+ // (V2ScanPartitioningAndOrdering is a separate early rule the rebuild does not run), so
+ // re-derive them on this single node. Safe on one node: the partitioning pass is idempotent
+ // and the ordering pass is applied once to a fresh node.
+ V2ScanPartitioningAndOrdering(scan).asInstanceOf[DataSourceV2ScanRelation]
+ }.filterNot { merged =>
+ // Decline if the merged scan degrades a partitioning/ordering an input reported -- that can
+ // force a shuffle/sort the original plan avoided -- unless the matching config opts in.
+ mergeDegradesReporting(merged, requiredKeyGroupedPartitioning, requiredOrdering)
Review Comment:
@-
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -700,17 +707,42 @@ class PlanMerger(
// order (npMapping.values would be exprId-hash-ordered).
val unionAttrs = cp.output ++ np.output.map(npMapping).filterNot(cp.outputSet.contains)
+ // The reported key-grouped partitioning / ordering the merged scan must preserve so BOTH inputs
+ // stay not-worse. Each input reports its own, remapped into cp's relation space (cp's already
+ // is; np's via npRelationMapping). The two usually agree (same table) but need not -- differing
+ // best-effort filters can prune different files, and a source may report per file set. Combine
+ // them into the single report the merge must keep (kGP: they must be equal; ordering: the
+ // stronger, which satisfies both). None from combine* means the inputs are INCOMPATIBLE -- no
+ // rebuilt scan could keep both not-worse -- so decline HERE, before rebuilding, unless the
+ // matching config accepts degrading that dimension.
+ val combinedKeyGroupedPartitioning = combineRequiredKeyGroupedPartitioning(
+ np.keyGroupedPartitioning.map(_.map(mapAttributes(_, npRelationMapping))).getOrElse(Nil),
+ cp.keyGroupedPartitioning.getOrElse(Nil))
+ val combinedOrdering = combineRequiredOrdering(
+ np.ordering.map(_.map(mapAttributes(_, npRelationMapping))).getOrElse(Nil),
+ cp.ordering.getOrElse(Nil))
+ if ((combinedKeyGroupedPartitioning.isEmpty && !dsv2AllowKeyGroupedPartitioningDegradation) ||
Review Comment:
@-
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -760,6 +794,16 @@ class PlanMerger(
// re-checks it), and the scan must produce exactly the requested union of columns.
strictFilters.forall(ExpressionSet(scan.pushedFilters).contains) &&
scan.outputSet == AttributeSet(unionAttrs)
+ }.map { scan =>
+ // rebuildScan returns the merged scan with reported partitioning/ordering unset
+ // (V2ScanPartitioningAndOrdering is a separate early rule the rebuild does not run), so
+ // re-derive them on this single node. Safe on one node: the partitioning pass is idempotent
+ // and the ordering pass is applied once to a fresh node.
+ V2ScanPartitioningAndOrdering(scan).asInstanceOf[DataSourceV2ScanRelation]
Review Comment:
@-
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3730359966
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
+ else None
+ }
+
+ // True when the rebuilt merged scan does not reproduce the required key-grouped partitioning, or
+ // does not satisfy the required ordering (the combined report the merge must preserve, computed
+ // at the leaf). Gated per dimension by the dsv2ScanMerge degradation configs; an empty required
+ // report imposes no constraint. Compared in cp's relation space.
+ private def mergeDegradesReporting(
Review Comment:
Agreed, and documented on the method: only the expressions are compared because that is all `DataSourceV2ScanRelation` carries. Worth noting it isn't specific to the `.orElse` either — the first build can prune differently from either input too.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
Review Comment:
Fixed, and this was the most valuable comment of the round. Both dimensions now compare via `isSameFunction`, recursively, so a nested transform's children are compared the same way. Note `BoundFunction.canonicalName()`'s default returns a random UUID, so this works exactly for connectors that override it — the same set for which SPJ works at all.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
Review Comment:
Fixed: the report check moved to the callers of `tryBuildMergedDSv2Scan` and runs per attempt, so a degradation no longer looks like a filter-enforcement failure. Separately, you're right that the retry itself has no test — deleting the `.orElse` leaves everything green. That predates this PR, so I've noted it as a follow-up rather than folding it in here.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
+ else None
+ }
+
+ // True when the rebuilt merged scan does not reproduce the required key-grouped partitioning, or
+ // does not satisfy the required ordering (the combined report the merge must preserve, computed
+ // at the leaf). Gated per dimension by the dsv2ScanMerge degradation configs; an empty required
+ // report imposes no constraint. Compared in cp's relation space.
+ private def mergeDegradesReporting(
+ merged: DataSourceV2ScanRelation,
+ requiredKeyGroupedPartitioning: Seq[Expression],
+ requiredOrdering: Seq[SortOrder]): Boolean = {
+ val kgpDegraded = !dsv2AllowKeyGroupedPartitioningDegradation &&
+ requiredKeyGroupedPartitioning.nonEmpty &&
+ !merged.keyGroupedPartitioning.exists(
+ _.map(_.canonicalized) == requiredKeyGroupedPartitioning.map(_.canonicalized))
+ val orderingDegraded = !dsv2AllowOrderingDegradation &&
+ requiredOrdering.nonEmpty &&
+ !SortOrder.orderingSatisfies(merged.ordering.getOrElse(Nil), requiredOrdering)
Review Comment:
Confirmed by measurement — `Nil` in place of `merged.ordering.getOrElse(Nil)` left all 82 tests green. Two new tests fix that. On the weaker-side variant: that is green only for two non-empty orderings; flipping the empty-side branches along with it fails the existing single-side decline test.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2618,6 +2619,109 @@ class MergeSubplansSuite extends PlanTest {
assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
}
+ test("SPARK-58549: do not merge DSv2 scans reporting incompatible kGP/ordering") {
Review Comment:
Second half confirmed and fixed — the config-on early-decline case now has a test. First half I couldn't reproduce: both mutations fail `DSv2PlanMergingSuite`'s end-to-end test, because the two subquery relations carry different exprIds, so the np-side remap is what makes the two reports comparable at all. They do stay green with `MergeSubplansSuite` alone.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3730362436
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7467,6 +7293,32 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val MERGE_SUBPLANS_DSV2_ALLOW_KEY_GROUPED_PARTITIONING_DEGRADATION = buildConf(
+ "spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowKeyGroupedPartitioningDegradation")
+ .doc("When false, a DataSource V2 scan merge is declined if the rebuilt merged scan would " +
Review Comment:
Fixed — both docs now say the flag also governs the decline that happens before any rebuild, when the two inputs' reports are incompatible.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -760,6 +794,16 @@ class PlanMerger(
// re-checks it), and the scan must produce exactly the requested union of columns.
strictFilters.forall(ExpressionSet(scan.pushedFilters).contains) &&
scan.outputSet == AttributeSet(unionAttrs)
+ }.map { scan =>
+ // rebuildScan returns the merged scan with reported partitioning/ordering unset
+ // (V2ScanPartitioningAndOrdering is a separate early rule the rebuild does not run), so
+ // re-derive them on this single node. Safe on one node: the partitioning pass is idempotent
+ // and the ordering pass is applied once to a fresh node.
+ V2ScanPartitioningAndOrdering(scan).asInstanceOf[DataSourceV2ScanRelation]
+ }.filterNot { merged =>
+ // Decline if the merged scan degrades a partitioning/ordering an input reported -- that can
+ // force a shuffle/sort the original plan avoided -- unless the matching config opts in.
+ mergeDegradesReporting(merged, requiredKeyGroupedPartitioning, requiredOrdering)
Review Comment:
Fixed by construction: the report check moved to the callers, so the build's `None` again means only that the strict filters cannot be re-enforced.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -700,17 +707,42 @@ class PlanMerger(
// order (npMapping.values would be exprId-hash-ordered).
val unionAttrs = cp.output ++ np.output.map(npMapping).filterNot(cp.outputSet.contains)
+ // The reported key-grouped partitioning / ordering the merged scan must preserve so BOTH inputs
+ // stay not-worse. Each input reports its own, remapped into cp's relation space (cp's already
+ // is; np's via npRelationMapping). The two usually agree (same table) but need not -- differing
+ // best-effort filters can prune different files, and a source may report per file set. Combine
+ // them into the single report the merge must keep (kGP: they must be equal; ordering: the
+ // stronger, which satisfies both). None from combine* means the inputs are INCOMPATIBLE -- no
+ // rebuilt scan could keep both not-worse -- so decline HERE, before rebuilding, unless the
+ // matching config accepts degrading that dimension.
+ val combinedKeyGroupedPartitioning = combineRequiredKeyGroupedPartitioning(
+ np.keyGroupedPartitioning.map(_.map(mapAttributes(_, npRelationMapping))).getOrElse(Nil),
+ cp.keyGroupedPartitioning.getOrElse(Nil))
+ val combinedOrdering = combineRequiredOrdering(
+ np.ordering.map(_.map(mapAttributes(_, npRelationMapping))).getOrElse(Nil),
+ cp.ordering.getOrElse(Nil))
+ if ((combinedKeyGroupedPartitioning.isEmpty && !dsv2AllowKeyGroupedPartitioningDegradation) ||
Review Comment:
Fixed, both here and in the `MergeContext` scaladoc. Two more turned up in the same pass: the `(Filter, Filter)` call site, and `DSv2DeferredScan`'s `@param`s.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -760,6 +794,16 @@ class PlanMerger(
// re-checks it), and the scan must produce exactly the requested union of columns.
strictFilters.forall(ExpressionSet(scan.pushedFilters).contains) &&
scan.outputSet == AttributeSet(unionAttrs)
+ }.map { scan =>
+ // rebuildScan returns the merged scan with reported partitioning/ordering unset
+ // (V2ScanPartitioningAndOrdering is a separate early rule the rebuild does not run), so
+ // re-derive them on this single node. Safe on one node: the partitioning pass is idempotent
+ // and the ordering pass is applied once to a fresh node.
+ V2ScanPartitioningAndOrdering(scan).asInstanceOf[DataSourceV2ScanRelation]
Review Comment:
The mechanism holds, but the two unmerged scans would each have carried the same `Some(Nil)` and collapsed the same way, so the merge isn't worse than not merging — the old `None` was accidentally better than the original plan. I've written that down in the comment rather than changing the behaviour.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
Review Comment:
Reworded. Your deeper point stands too: differing pruning explains differing split counts and partition values, not differing key expressions, and only the expressions are compared — so the comment now points at the ordering dimension, where a per-scan report really can arise.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #57753:
URL: https://github.com/apache/spark/pull/57753#issuecomment-5182777690
Thanks @dongjoon-hyun! Both applied — the `SPARK-58549` test prefixes in 4016674883f, and `requiredKeyGroupedPartitioning`/`requiredOrdering` renamed to `combined*` at the `tryMergeDSv2` use site. Good catch on the double meaning: the `Option`s now read as "the combined report, None if incompatible", and the `Seq` parameters keep `required*` where empty means no requirement.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3714577409
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -700,17 +707,42 @@ class PlanMerger(
// order (npMapping.values would be exprId-hash-ordered).
val unionAttrs = cp.output ++ np.output.map(npMapping).filterNot(cp.outputSet.contains)
+ // The reported key-grouped partitioning / ordering the merged scan must preserve so BOTH inputs
+ // stay not-worse. Each input reports its own, remapped into cp's relation space (cp's already
+ // is; np's via npRelationMapping). The two usually agree (same table) but need not -- differing
+ // best-effort filters can prune different files, and a source may report per file set. Combine
+ // them into the single report the merge must keep (kGP: they must be equal; ordering: the
+ // stronger, which satisfies both). None from combine* means the inputs are INCOMPATIBLE -- no
+ // rebuilt scan could keep both not-worse -- so decline HERE, before rebuilding, unless the
+ // matching config accepts degrading that dimension.
+ val requiredKeyGroupedPartitioning = combineRequiredKeyGroupedPartitioning(
Review Comment:
`requiredKeyGroupedPartitioning` / `requiredOrdering` seem to carry a double meaning here. These `Option`s use `.isEmpty` to mean the inputs' reports are *incompatible*, while the same names are reused for the `Seq` parameters of `tryBuildMergedDSv2Scan` / `mergeDegradesReporting`, where empty means *no requirement*. Renaming the `Option`s to `combined*` (and deriving the `expected*` `Seq`s from them) makes the early-decline condition read correctly at the use site. Maybe
- `requiredKeyGroupedPartitioning` -> `combinedKeyGroupedPartitioning`
- `requiredOrdering` -> `combinedOrdering`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3730364700
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2600,10 +2600,11 @@ class MergeSubplansSuite extends PlanTest {
}
test("SPARK-40259: do not merge DSv2 scans that report key-grouped partitioning or ordering") {
- // The rebuilt merged scan does not reconstruct reported partitioning/ordering, so a scan
- // reporting either declines the merge (checked on both the np and cp side) -- the plan is left
- // unchanged -- rather than silently dropping it. Preserving them across a merge is a deferred
- // follow-up. (The plain-scan merge is already covered by the projected-columns test above.)
+ // An input reports key-grouped partitioning or ordering, but the merged scan -- rebuilt over a
Review Comment:
Renamed to "do not merge DSv2 scans when the merge would degrade a reported key-grouped partitioning or ordering". Kept the `SPARK-40259` prefix, since the test predates this PR.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala:
##########
@@ -169,4 +172,39 @@ class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession
}
}
}
+
+ test("SPARK-58549: a scan merge preserves the sources' reported key-grouped partitioning") {
+ val t = "scanmergereport.t2"
+ withTable(t) {
+ withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "true") {
Review Comment:
Correct. Rather than drop it I made it matter: the test now also asserts the merged scan reports `KeyedPartitioning` on `c1` in the physical plan, with AQE off so `executedPlan` is walkable.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7467,6 +7293,32 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val MERGE_SUBPLANS_DSV2_ALLOW_KEY_GROUPED_PARTITIONING_DEGRADATION = buildConf(
+ "spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowKeyGroupedPartitioningDegradation")
Review Comment:
Fixed. I dropped `allow` at the same time — with it, the key came to 94 chars, one over the line limit, which would have made it the only key in `SQLConf` split across two literals and so unsearchable from a log.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryScanMergingPartitionFilterTable.scala:
##########
@@ -82,11 +83,57 @@ class InMemoryScanMergingPartitionFilterTable(
* Thin scan decorator that exposes only `readSchema`, `toBatch` and `description`, dropping the
* base scan's `SupportsReportPartitioning`/`SupportsReportStatistics`. So the scan relation carries
* no reported partitioning/ordering/statistics -- for a partitioned table this keeps
- * `keyGroupedPartitioning` unset, which the scan merge requires (preserving reported partitioning
- * across a merge is a separate follow-up).
+ * `keyGroupedPartitioning` unset, so the fixture stays focused on pushdown; preserving reported
+ * partitioning across a merge is exercised by [[InMemoryScanMergingReportingTable]].
*/
case class NonReportingScan(inner: Scan) extends Scan {
override def readSchema(): StructType = inner.readSchema()
override def toBatch: Batch = inner.toBatch
override def description(): String = inner.description()
}
+
+/**
+ * Like [[InMemoryScanMergingPartitionFilterCatalog]] but hands out tables that KEEP their reported
+ * partitioning/ordering (no [[NonReportingScan]] wrapper), so a scan merge that must preserve the
+ * reported key-grouped partitioning across the merge can be exercised.
+ */
+class InMemoryScanMergingReportingCatalog
+ extends InMemoryTableEnhancedPartitionFilterCatalog {
+ import CatalogV2Implicits._
+
+ override def createTable(
Review Comment:
Fixed: a `newScanMergingTable` hook on the catalog, and `InMemoryScanMergingPartitionFilterTable` now extends the reporting table, so `createTable` and `capabilities()` each exist once.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #57753:
URL: https://github.com/apache/spark/pull/57753#issuecomment-5230990433
Thank you @uros-b, @dongjoon-hyun and @LuciferYang for the review!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #57753:
URL: https://github.com/apache/spark/pull/57753#issuecomment-5180855594
cc @LuciferYang
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #57753:
URL: https://github.com/apache/spark/pull/57753#issuecomment-5181396784
I’ll take a look at it tomorrow.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3730359805
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
+ else None
+ }
+
+ // True when the rebuilt merged scan does not reproduce the required key-grouped partitioning, or
+ // does not satisfy the required ordering (the combined report the merge must preserve, computed
+ // at the leaf). Gated per dimension by the dsv2ScanMerge degradation configs; an empty required
+ // report imposes no constraint. Compared in cp's relation space.
+ private def mergeDegradesReporting(
+ merged: DataSourceV2ScanRelation,
+ requiredKeyGroupedPartitioning: Seq[Expression],
+ requiredOrdering: Seq[SortOrder]): Boolean = {
+ val kgpDegraded = !dsv2AllowKeyGroupedPartitioningDegradation &&
+ requiredKeyGroupedPartitioning.nonEmpty &&
Review Comment:
@-
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
+ else None
+ }
+
+ // True when the rebuilt merged scan does not reproduce the required key-grouped partitioning, or
+ // does not satisfy the required ordering (the combined report the merge must preserve, computed
+ // at the leaf). Gated per dimension by the dsv2ScanMerge degradation configs; an empty required
+ // report imposes no constraint. Compared in cp's relation space.
+ private def mergeDegradesReporting(
Review Comment:
@-
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
Review Comment:
@-
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
Review Comment:
@-
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
+ else None
+ }
+
+ // True when the rebuilt merged scan does not reproduce the required key-grouped partitioning, or
+ // does not satisfy the required ordering (the combined report the merge must preserve, computed
+ // at the leaf). Gated per dimension by the dsv2ScanMerge degradation configs; an empty required
+ // report imposes no constraint. Compared in cp's relation space.
+ private def mergeDegradesReporting(
+ merged: DataSourceV2ScanRelation,
+ requiredKeyGroupedPartitioning: Seq[Expression],
+ requiredOrdering: Seq[SortOrder]): Boolean = {
+ val kgpDegraded = !dsv2AllowKeyGroupedPartitioningDegradation &&
+ requiredKeyGroupedPartitioning.nonEmpty &&
+ !merged.keyGroupedPartitioning.exists(
+ _.map(_.canonicalized) == requiredKeyGroupedPartitioning.map(_.canonicalized))
+ val orderingDegraded = !dsv2AllowOrderingDegradation &&
+ requiredOrdering.nonEmpty &&
+ !SortOrder.orderingSatisfies(merged.ordering.getOrElse(Nil), requiredOrdering)
Review Comment:
@-
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3714501965
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala:
##########
@@ -169,4 +172,39 @@ class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession
}
}
}
+
+ test("SPARK-40259: a scan merge preserves the sources' reported key-grouped partitioning") {
Review Comment:
```suggestion
test("SPARK-58549: a scan merge preserves the sources' reported key-grouped partitioning") {
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2618,6 +2619,109 @@ class MergeSubplansSuite extends PlanTest {
assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
}
+ test("SPARK-40259: do not merge DSv2 scans reporting incompatible kGP/ordering") {
Review Comment:
```suggestion
test("SPARK-58549: do not merge DSv2 scans reporting incompatible kGP/ordering") {
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2618,6 +2619,109 @@ class MergeSubplansSuite extends PlanTest {
assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
}
+ test("SPARK-40259: do not merge DSv2 scans reporting incompatible kGP/ordering") {
+ // Both inputs report a partitioning/ordering, but on the different column each reads, so no
+ // single rebuilt scan could keep both not-worse. combineRequired* returns None (incompatible),
+ // so the merge is declined at the leaf -- before any rebuild -- with the degradation configs
+ // off (the default). This is distinct from the single-side case above (which rebuilds and then
+ // finds the re-derived report degraded): here the two inputs disagree with each other up front.
+ def assertDeclines(withField: DataSourceV2ScanRelation => DataSourceV2ScanRelation): Unit = {
+ // withField applied to the "a" scan reports on a, applied to the "b" scan reports on b.
+ val q = testRelation.select(
+ ScalarSubquery(withField(v2ScanReading("a")).groupBy()(sum($"a").as("sa"))),
+ ScalarSubquery(withField(v2ScanReading("b")).groupBy()(sum($"b").as("sb"))))
+ comparePlans(Optimize.execute(q.analyze), q.analyze)
+ }
+
+ assertDeclines(s => s.copy(keyGroupedPartitioning = Some(Seq(s.output.head))))
+ assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
+ }
+
+ test("SPARK-40259: merge DSv2 scans reporting kGP/ordering when the degradation config allows") {
Review Comment:
```suggestion
test("SPARK-58549: merge DSv2 scans reporting kGP/ordering when the degradation config allows") {
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2618,6 +2619,109 @@ class MergeSubplansSuite extends PlanTest {
assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
}
+ test("SPARK-40259: do not merge DSv2 scans reporting incompatible kGP/ordering") {
+ // Both inputs report a partitioning/ordering, but on the different column each reads, so no
+ // single rebuilt scan could keep both not-worse. combineRequired* returns None (incompatible),
+ // so the merge is declined at the leaf -- before any rebuild -- with the degradation configs
+ // off (the default). This is distinct from the single-side case above (which rebuilds and then
+ // finds the re-derived report degraded): here the two inputs disagree with each other up front.
+ def assertDeclines(withField: DataSourceV2ScanRelation => DataSourceV2ScanRelation): Unit = {
+ // withField applied to the "a" scan reports on a, applied to the "b" scan reports on b.
+ val q = testRelation.select(
+ ScalarSubquery(withField(v2ScanReading("a")).groupBy()(sum($"a").as("sa"))),
+ ScalarSubquery(withField(v2ScanReading("b")).groupBy()(sum($"b").as("sb"))))
+ comparePlans(Optimize.execute(q.analyze), q.analyze)
+ }
+
+ assertDeclines(s => s.copy(keyGroupedPartitioning = Some(Seq(s.output.head))))
+ assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
+ }
+
+ test("SPARK-40259: merge DSv2 scans reporting kGP/ordering when the degradation config allows") {
+ // With the matching degradation config on, a merge that would drop a reported partitioning or
+ // ordering proceeds anyway (trading it for a single scan). The merged scan re-derives no report
+ // from the non-reporting TestV2Scan, so the fused plan is the plain column union.
+ val mergedScan = v2ScanReadingOn(v2Table, Seq("a", "b"))
+ val mergedSubquery = mergedScan
+ .groupBy()(sum($"a").as("sum_a"), sum($"b").as("sum_b"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("sum_b"), $"sum_b")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ def assertMerges(
+ withField: DataSourceV2ScanRelation => DataSourceV2ScanRelation, confKey: String): Unit = {
+ val sub1 = ScalarSubquery(withField(v2ScanReading("a")).groupBy()(sum($"a").as("sum_a")))
+ val sub2 = ScalarSubquery(v2ScanReading("b").groupBy()(sum($"b").as("sum_b")))
+ val originalQuery = testRelation.select(sub1, sub2)
+ withSQLConf(confKey -> "true") {
+ comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+ }
+ }
+
+ assertMerges(s => s.copy(keyGroupedPartitioning = Some(Seq(s.output.head))),
+ SQLConf.MERGE_SUBPLANS_DSV2_ALLOW_KEY_GROUPED_PARTITIONING_DEGRADATION.key)
+ assertMerges(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))),
+ SQLConf.MERGE_SUBPLANS_DSV2_ALLOW_ORDERING_DEGRADATION.key)
+ }
+
+ test("SPARK-40259: enforce the required report on the deferred under-Filter scan build") {
Review Comment:
```suggestion
test("SPARK-58549: enforce the required report on the deferred under-Filter scan build") {
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3730359805
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
+ else None
+ }
+
+ // True when the rebuilt merged scan does not reproduce the required key-grouped partitioning, or
+ // does not satisfy the required ordering (the combined report the merge must preserve, computed
+ // at the leaf). Gated per dimension by the dsv2ScanMerge degradation configs; an empty required
+ // report imposes no constraint. Compared in cp's relation space.
+ private def mergeDegradesReporting(
+ merged: DataSourceV2ScanRelation,
+ requiredKeyGroupedPartitioning: Seq[Expression],
+ requiredOrdering: Seq[SortOrder]): Boolean = {
+ val kgpDegraded = !dsv2AllowKeyGroupedPartitioningDegradation &&
+ requiredKeyGroupedPartitioning.nonEmpty &&
Review Comment:
Both halves are answered in the summary comment — short version: the gain is real and is now documented as deliberate, and emptying the requirement when the inputs conflict is a no-op, because that dimension's config being on already short-circuits `mergeDegradesReporting`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #57753:
URL: https://github.com/apache/spark/pull/57753#issuecomment-5207280556
No problem at all, and thanks — this review was greatly appreciated. It found real gaps, including a couple I'd have shipped. Everything is in `97d5247`.
**What changed**
- **Transform comparison (`:858`).** This was the big one. `TransformExpression` is a plain case class, so `canonicalized ==` compares the `BoundFunction` instance, and `V2ExpressionUtils` binds afresh on every derivation — so two identical `bucket(4, id)` reports compared unequal and every transform-partitioned source declined the merge. Both dimensions now compare with `isSameFunction`, recursively, so a nested transform's children don't fall back to the instance comparison either. One fixture trap worth flagging: Spark's `UnboundBucketFunction.bind` returns a singleton, so an in-repo bucket test passes with or without the fix — the new test uses a `FunctionCatalog` that binds a fresh instance per call, like a real connector, and it fails without the fix.
- **Report checked per build attempt (`:831`, `:806`).** The check moved out of `tryBuildMergedDSv2Scan` to its callers, and now runs per attempt. A degradation is no longer indistinguishable from a filter-enforcement failure, so the build's `None` again means only "strict filters not re-enforceable" — which restores the `tryBuildFilterDSv2ScanChild` contract you flagged — and a source that reports per scan still gets its chance from the strict-only attempt.
- **Coverage (`:877`, `:2622`).** You were right that the ordering check never fired with a non-empty requirement; I measured it — replacing `merged.ordering.getOrElse(Nil)` with `Nil` left all 82 tests green. `TestV2Table` can now report an ordering or a bucket partitioning, and there are four new tests: the two ordering ones, the bucket-transform one, and the incompatible-inputs-with-the-config-on case, which was indeed uncovered.
- **Configs (`:7297`, `:7298`).** Both renamed to end in `.enabled`, and both docs now mention the decline that happens before any rebuild. I dropped the redundant `allow` while renaming, because `allowKeyGroupedPartitioningDegradation.enabled` came to 94 chars — one over the line limit, so it would have been the only key in `SQLConf` split across two literals, and not greppable from a log.
- **Comments and names (`:724`, `:841`, `:2603`).** All fixed, plus two more the same pass turned up: the `(Filter, Filter)` call site still claimed strict filters were the only reason to decline, and `DSv2DeferredScan`'s `@param`s named the configs by their old keys.
- **Nits (`:179`, `:104`).** The `withSQLConf` was decorative as you say — rather than drop it I made it matter, by also asserting the merged scan reports `KeyedPartitioning` on the partition column in the physical plan (AQE off so the plan is walkable). The fixture is down to one `createTable` and one `capabilities()`.
**Where I'd push back a little**
- `:872`, second half — emptying the requirement at line 730 is a no-op rather than a lost check: reaching `getOrElse(Nil)` with a `None` implies that dimension's config is on, and `mergeDegradesReporting` already short-circuits on it.
- `:872`, first half — a *gained* report is real (the partitioning pass is reference-subset guarded, so `{c1,c3}` + `{c2,c3}` does give the merged scan `Some([c1,c2])`), but I read it as the win rather than a hazard: the source reported that partitioning for the table, and an input dropped it only because a pruned column left it inexpressible. I've documented it so it reads as deliberate.
- `:867` — agreed the check sees only the expressions, and that limitation is now written down. `keyGroupedPartitioning` is `Option[Seq[Expression]]`, so split counts and partition values aren't visible at that layer, and planning input partitions in the optimizer to compare them seems worse than the gap. It isn't specific to the `.orElse` either — the first build can prune differently from either input too.
- `:802` — a zero-key `KeyGroupedPartitioning` does now reach the merged scan as `Some(Nil)`. But the two unmerged scans would each have carried the same `Some(Nil)` and collapsed the same way, so it's not worse than not merging — the old `None` was accidentally better than the original plan. Commented rather than changed.
- `:2622`, first half — I couldn't reproduce this one: setting the equality to `false`, and deleting the np-side `mapAttributes`, both fail `DSv2PlanMergingSuite`'s new end-to-end test. The two subquery relations carry different exprIds, so the remap is what makes the two reports comparable at all. Both do stay green if `MergeSubplansSuite` runs alone.
- `:877`, second half — picking the weaker side is green only for two non-empty orderings; flipping the empty-side branches along with it fails the existing single-side decline test. The new test covers the non-empty case.
**Two follow-ups I'd rather not fold in here**
- `TransformExpression`'s equality is the root cause and it reaches past this rule: `PartitioningPreservingUnaryExecNode.projectKeyedPartitionings` puts kGP expressions in an `ExpressionSet` (so dedup silently fails and the cross-product can hit `aliasCandidateLimit`), `BatchScanExec.equals` compares `keyGroupedPartitioning` directly, and `DataSourceV2ScanRelation`'s canonical form does too, so `PlanMerger.checkIdenticalPlans` can miss two identical subqueries. Overriding `equals`/`hashCode` in terms of `canonicalName()` would fix all of them and let me delete the helper this PR adds. I'll file a JIRA.
- The strict-only retry in `tryBuildFilterDSv2ScanChild` is untested — deleting the `.orElse` leaves everything green — because the fixture is all-or-nothing on filters, so both attempts fail together. That predates this PR (SPARK-40259); it needs a source that pushes a batch or nothing.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3730364539
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
Review Comment:
@-
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2600,10 +2600,11 @@ class MergeSubplansSuite extends PlanTest {
}
test("SPARK-40259: do not merge DSv2 scans that report key-grouped partitioning or ordering") {
- // The rebuilt merged scan does not reconstruct reported partitioning/ordering, so a scan
- // reporting either declines the merge (checked on both the np and cp side) -- the plan is left
- // unchanged -- rather than silently dropping it. Preserving them across a merge is a deferred
- // follow-up. (The plain-scan merge is already covered by the projected-columns test above.)
+ // An input reports key-grouped partitioning or ordering, but the merged scan -- rebuilt over a
Review Comment:
@-
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala:
##########
@@ -169,4 +172,39 @@ class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession
}
}
}
+
+ test("SPARK-58549: a scan merge preserves the sources' reported key-grouped partitioning") {
+ val t = "scanmergereport.t2"
+ withTable(t) {
+ withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "true") {
Review Comment:
@-
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7467,6 +7293,32 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val MERGE_SUBPLANS_DSV2_ALLOW_KEY_GROUPED_PARTITIONING_DEGRADATION = buildConf(
+ "spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowKeyGroupedPartitioningDegradation")
Review Comment:
@-
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryScanMergingPartitionFilterTable.scala:
##########
@@ -82,11 +83,57 @@ class InMemoryScanMergingPartitionFilterTable(
* Thin scan decorator that exposes only `readSchema`, `toBatch` and `description`, dropping the
* base scan's `SupportsReportPartitioning`/`SupportsReportStatistics`. So the scan relation carries
* no reported partitioning/ordering/statistics -- for a partitioned table this keeps
- * `keyGroupedPartitioning` unset, which the scan merge requires (preserving reported partitioning
- * across a merge is a separate follow-up).
+ * `keyGroupedPartitioning` unset, so the fixture stays focused on pushdown; preserving reported
+ * partitioning across a merge is exercised by [[InMemoryScanMergingReportingTable]].
*/
case class NonReportingScan(inner: Scan) extends Scan {
override def readSchema(): StructType = inner.readSchema()
override def toBatch: Batch = inner.toBatch
override def description(): String = inner.description()
}
+
+/**
+ * Like [[InMemoryScanMergingPartitionFilterCatalog]] but hands out tables that KEEP their reported
+ * partitioning/ordering (no [[NonReportingScan]] wrapper), so a scan merge that must preserve the
+ * reported key-grouped partitioning across the merge can be exercised.
+ */
+class InMemoryScanMergingReportingCatalog
+ extends InMemoryTableEnhancedPartitionFilterCatalog {
+ import CatalogV2Implicits._
+
+ override def createTable(
Review Comment:
@-
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3726033676
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
+ else None
+ }
+
+ // True when the rebuilt merged scan does not reproduce the required key-grouped partitioning, or
+ // does not satisfy the required ordering (the combined report the merge must preserve, computed
+ // at the leaf). Gated per dimension by the dsv2ScanMerge degradation configs; an empty required
+ // report imposes no constraint. Compared in cp's relation space.
+ private def mergeDegradesReporting(
+ merged: DataSourceV2ScanRelation,
+ requiredKeyGroupedPartitioning: Seq[Expression],
+ requiredOrdering: Seq[SortOrder]): Boolean = {
+ val kgpDegraded = !dsv2AllowKeyGroupedPartitioningDegradation &&
+ requiredKeyGroupedPartitioning.nonEmpty &&
Review Comment:
The not-worse check catches a report weaker than an input's, not one that appears when neither input had one.
- on a `(c1,c2)` table with the sides reading `{c1,c3}` and `{c2,c3}`, the merged scan gains `Some([c1,c2])` and starts taking the `KeyedPartitioning` branch of `replanWithRuntimeFilters`: pruned splits get padded with `None` and launch empty tasks, and three `SparkException` invariants now apply.
- when the inputs conflict and the config is on, line 730 empties the requirement, so that dimension's post-rebuild check goes too.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
+ else None
+ }
+
+ // True when the rebuilt merged scan does not reproduce the required key-grouped partitioning, or
+ // does not satisfy the required ordering (the combined report the merge must preserve, computed
+ // at the leaf). Gated per dimension by the dsv2ScanMerge degradation configs; an empty required
+ // report imposes no constraint. Compared in cp's relation space.
+ private def mergeDegradesReporting(
+ merged: DataSourceV2ScanRelation,
+ requiredKeyGroupedPartitioning: Seq[Expression],
+ requiredOrdering: Seq[SortOrder]): Boolean = {
+ val kgpDegraded = !dsv2AllowKeyGroupedPartitioningDegradation &&
+ requiredKeyGroupedPartitioning.nonEmpty &&
+ !merged.keyGroupedPartitioning.exists(
+ _.map(_.canonicalized) == requiredKeyGroupedPartitioning.map(_.canonicalized))
+ val orderingDegraded = !dsv2AllowOrderingDegradation &&
+ requiredOrdering.nonEmpty &&
+ !SortOrder.orderingSatisfies(merged.ordering.getOrElse(Nil), requiredOrdering)
Review Comment:
No test makes the merged scan carry a non-empty `ordering`, so the check at 877 never returns false with a non-empty requirement. replacing `merged.ordering.getOrElse(Nil)` with `Nil` stays green, and that switches merging off for every `SupportsReportOrdering` source. making `combineRequiredOrdering` pick the weaker side is also green, and would combine `[a]` and `[a, b]` into `[a]`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
Review Comment:
The ordering dimension has kGP's root cause: `orderingSatisfies` goes to `SortOrder.satisfies`, then `semanticEquals`, which is `canonicalized ==`. for a source reporting `sort(bucket(4, id))` the two binds differ, so both sides report the same ordering and the pair is still declined at the leaf.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
Review Comment:
The comment above `combineRequiredKeyGroupedPartitioning` says kGP "must be equal (bucketing is a table property)", and on that reasoning `else None` is dead code. Line 712 says the two sides "usually agree ... but need not" and lists the reasons.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2600,10 +2600,11 @@ class MergeSubplansSuite extends PlanTest {
}
test("SPARK-40259: do not merge DSv2 scans that report key-grouped partitioning or ordering") {
- // The rebuilt merged scan does not reconstruct reported partitioning/ordering, so a scan
- // reporting either declines the merge (checked on both the np and cp side) -- the plan is left
- // unchanged -- rather than silently dropping it. Preserving them across a merge is a deferred
- // follow-up. (The plain-scan merge is already covered by the projected-columns test above.)
+ // An input reports key-grouped partitioning or ordering, but the merged scan -- rebuilt over a
Review Comment:
The test is still named `do not merge DSv2 scans that report key-grouped partitioning or ordering`, and this PR removes that rule; the comment underneath now says the merge would degrade what the input reported.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
Review Comment:
The report-related declines moved from the read-only gate to behind `rebuildScan`, so each attempt costs one or two full rebuilds, and `merge` walks the whole cache per subplan.
- connector that must list files to answer `numPartitions()` pays file planning at optimizer time.
- the `.orElse` cannot tell the two `None` causes apart. A structural degradation fails again; a source reporting per pruned file set makes the retry succeed instead.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -760,6 +794,16 @@ class PlanMerger(
// re-checks it), and the scan must produce exactly the requested union of columns.
strictFilters.forall(ExpressionSet(scan.pushedFilters).contains) &&
scan.outputSet == AttributeSet(unionAttrs)
+ }.map { scan =>
+ // rebuildScan returns the merged scan with reported partitioning/ordering unset
+ // (V2ScanPartitioningAndOrdering is a separate early rule the rebuild does not run), so
+ // re-derive them on this single node. Safe on one node: the partitioning pass is idempotent
+ // and the ordering pass is applied once to a fresh node.
+ V2ScanPartitioningAndOrdering(scan).asInstanceOf[DataSourceV2ScanRelation]
Review Comment:
The change keeps merging when both sides report `Some(Nil)` and also has the merged scan re-derive its own report, so `Some(Nil)` now lands on the merged scan where it used to be `None`. For a source reporting a zero-key `KeyGroupedPartitioning` with `HasPartitionKey` splits, `groupedSatisfies` is trivially true over the empty set, so a `GroupPartitionsExec` collapses every split into one task.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -784,14 +828,56 @@ class PlanMerger(
// turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
// rather than carrying it on DSv2DeferredScan.
child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
- tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
- .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None,
+ d.requiredKeyGroupedPartitioning, d.requiredOrdering))
.map { built =>
child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
}
}
}
+ // The key-grouped partitioning the merged scan must reproduce to keep both inputs not-worse: they
+ // must be equal (bucketing is a table property), so a differing non-empty pair is INCOMPATIBLE
+ // (None); an empty side imposes no constraint. Compared canonically in cp's relation space (np's
+ // report was remapped into it by the caller).
+ private def combineRequiredKeyGroupedPartitioning(
+ a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
+ if (a.isEmpty) Some(b)
+ else if (b.isEmpty) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
+ else None
+ }
+
+ // The ordering the merged scan must satisfy to keep both inputs not-worse: the stronger of the
+ // two (the one that satisfies the other -- satisfying it implies satisfying the weaker). If
+ // neither satisfies the other they are INCOMPATIBLE (None). An empty ordering never constrains.
+ private def combineRequiredOrdering(
+ a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
+ else None
+ }
+
+ // True when the rebuilt merged scan does not reproduce the required key-grouped partitioning, or
+ // does not satisfy the required ordering (the combined report the merge must preserve, computed
+ // at the leaf). Gated per dimension by the dsv2ScanMerge degradation configs; an empty required
+ // report imposes no constraint. Compared in cp's relation space.
+ private def mergeDegradesReporting(
Review Comment:
`mergeDegradesReporting` compares the expression list only, but SPJ also needs `numPartitions` and `partitionKeys` to match. once the `.orElse` at 831 drops the best-effort filter the split count grows, so the expression list survives while the key set changes. The check still returns false, an SPJ join no longer lines up, and the shuffle comes back.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala:
##########
@@ -2618,6 +2619,109 @@ class MergeSubplansSuite extends PlanTest {
assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
}
+ test("SPARK-58549: do not merge DSv2 scans reporting incompatible kGP/ordering") {
Review Comment:
1. `TestV2Scan` implements only `Scan`, so the rebuilt merged scan re-derives no report and all three new tests collapse to "any non-empty requirement fails". setting the equality at 849 to `false` stays green, and so does deleting the np-side `mapAttributes` at 718/722.
2. The "incompatible kGP/ordering" test runs the default configs only, so the config terms in the early decline at 724-726 have no coverage. changing it to `if (combinedKgp.isEmpty || combinedOrdering.isEmpty)` keeps the suite and the end-to-end test green, and that is the scenario the config doc promises. a change to which side's report survives a conflict has nothing watching it either, since emptying the requirement disables the post-rebuild check.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -700,17 +707,42 @@ class PlanMerger(
// order (npMapping.values would be exprId-hash-ordered).
val unionAttrs = cp.output ++ np.output.map(npMapping).filterNot(cp.outputSet.contains)
+ // The reported key-grouped partitioning / ordering the merged scan must preserve so BOTH inputs
+ // stay not-worse. Each input reports its own, remapped into cp's relation space (cp's already
+ // is; np's via npRelationMapping). The two usually agree (same table) but need not -- differing
+ // best-effort filters can prune different files, and a source may report per file set. Combine
+ // them into the single report the merge must keep (kGP: they must be equal; ordering: the
+ // stronger, which satisfies both). None from combine* means the inputs are INCOMPATIBLE -- no
+ // rebuilt scan could keep both not-worse -- so decline HERE, before rebuilding, unless the
+ // matching config accepts degrading that dimension.
+ val combinedKeyGroupedPartitioning = combineRequiredKeyGroupedPartitioning(
+ np.keyGroupedPartitioning.map(_.map(mapAttributes(_, npRelationMapping))).getOrElse(Nil),
+ cp.keyGroupedPartitioning.getOrElse(Nil))
+ val combinedOrdering = combineRequiredOrdering(
+ np.ordering.map(_.map(mapAttributes(_, npRelationMapping))).getOrElse(Nil),
+ cp.ordering.getOrElse(Nil))
+ if ((combinedKeyGroupedPartitioning.isEmpty && !dsv2AllowKeyGroupedPartitioningDegradation) ||
Review Comment:
Line 699's "Eligibility is settled above; everything below constructs the merge" and the matching sentence in the `MergeContext` scaladoc no longer hold on the deferred path.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -760,6 +794,16 @@ class PlanMerger(
// re-checks it), and the scan must produce exactly the requested union of columns.
strictFilters.forall(ExpressionSet(scan.pushedFilters).contains) &&
scan.outputSet == AttributeSet(unionAttrs)
+ }.map { scan =>
+ // rebuildScan returns the merged scan with reported partitioning/ordering unset
+ // (V2ScanPartitioningAndOrdering is a separate early rule the rebuild does not run), so
+ // re-derive them on this single node. Safe on one node: the partitioning pass is idempotent
+ // and the ordering pass is applied once to a fresh node.
+ V2ScanPartitioningAndOrdering(scan).asInstanceOf[DataSourceV2ScanRelation]
+ }.filterNot { merged =>
+ // Decline if the merged scan degrades a partitioning/ordering an input reported -- that can
+ // force a shuffle/sort the original plan avoided -- unless the matching config opts in.
+ mergeDegradesReporting(merged, requiredKeyGroupedPartitioning, requiredOrdering)
Review Comment:
The scaladoc on `tryBuildFilterDSv2ScanChild` says `None` comes back only when the strict filters cannot be re-enforced. Line 806 gives `tryBuildMergedDSv2Scan` a second source of `None`, so that no longer holds.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryScanMergingPartitionFilterTable.scala:
##########
@@ -82,11 +83,57 @@ class InMemoryScanMergingPartitionFilterTable(
* Thin scan decorator that exposes only `readSchema`, `toBatch` and `description`, dropping the
* base scan's `SupportsReportPartitioning`/`SupportsReportStatistics`. So the scan relation carries
* no reported partitioning/ordering/statistics -- for a partitioned table this keeps
- * `keyGroupedPartitioning` unset, which the scan merge requires (preserving reported partitioning
- * across a merge is a separate follow-up).
+ * `keyGroupedPartitioning` unset, so the fixture stays focused on pushdown; preserving reported
+ * partitioning across a merge is exercised by [[InMemoryScanMergingReportingTable]].
*/
case class NonReportingScan(inner: Scan) extends Scan {
override def readSchema(): StructType = inner.readSchema()
override def toBatch: Batch = inner.toBatch
override def description(): String = inner.description()
}
+
+/**
+ * Like [[InMemoryScanMergingPartitionFilterCatalog]] but hands out tables that KEEP their reported
+ * partitioning/ordering (no [[NonReportingScan]] wrapper), so a scan merge that must preserve the
+ * reported key-grouped partitioning across the merge can be exercised.
+ */
+class InMemoryScanMergingReportingCatalog
+ extends InMemoryTableEnhancedPartitionFilterCatalog {
+ import CatalogV2Implicits._
+
+ override def createTable(
Review Comment:
nit: `InMemoryScanMergingReportingCatalog.createTable` is the third copy of this body (the parent plus the sibling in this file), and `capabilities()` is copied verbatim too.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7467,6 +7293,32 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val MERGE_SUBPLANS_DSV2_ALLOW_KEY_GROUPED_PARTITIONING_DEGRADATION = buildConf(
+ "spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowKeyGroupedPartitioningDegradation")
+ .doc("When false, a DataSource V2 scan merge is declined if the rebuilt merged scan would " +
Review Comment:
The two `.doc()` strings only mention "if the rebuilt merged scan would report weaker ...", but the same flag also governs the early decline at 724-727, which tests whether the two inputs' reports are incompatible, before any rebuild.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala:
##########
@@ -169,4 +172,39 @@ class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession
}
}
}
+
+ test("SPARK-58549: a scan merge preserves the sources' reported key-grouped partitioning") {
+ val t = "scanmergereport.t2"
+ withTable(t) {
+ withSQLConf(SQLConf.V2_BUCKETING_ENABLED.key -> "true") {
Review Comment:
nit: The `withSQLConf(V2_BUCKETING_ENABLED -> "true")` here does nothing: it is true by default, only the physical `outputPartitioning` reads it, and every assertion is on `optimizedPlan`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7467,6 +7293,32 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val MERGE_SUBPLANS_DSV2_ALLOW_KEY_GROUPED_PARTITIONING_DEGRADATION = buildConf(
+ "spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowKeyGroupedPartitioningDegradation")
Review Comment:
These two open a new `mergeSubplans.dsv2ScanMerge.*` namespace and are the only booleans in the family without an `.enabled` suffix; the existing four sit under `filterPropagation` and all end in `.enabled`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth closed pull request #57753: [SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge
URL: https://github.com/apache/spark/pull/57753
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org