peter-toth commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3730364539
##########
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:
@-
##########
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:
@-
##########
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:
@-
##########
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:
@-
##########
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:
@-
--
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]
