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