peter-toth commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3730364700
##########
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:
Renamed to "do not merge DSv2 scans when the merge would degrade a reported
key-grouped partitioning or ordering". Kept the `SPARK-40259` prefix, since the
test predates this PR.
##########
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:
Correct. Rather than drop it I made it matter: the test now also asserts the
merged scan reports `KeyedPartitioning` on `c1` in the physical plan, with AQE
off so `executedPlan` is walkable.
##########
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:
Fixed. I dropped `allow` at the same time — with it, the key came to 94
chars, one over the line limit, which would have made it the only key in
`SQLConf` split across two literals and so unsearchable from a log.
##########
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:
Fixed: a `newScanMergingTable` hook on the catalog, and
`InMemoryScanMergingPartitionFilterTable` now extends the reporting table, so
`createTable` and `capabilities()` each exist once.
--
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]
