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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
