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