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