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