dongjoon-hyun commented on code in PR #57753:
URL: https://github.com/apache/spark/pull/57753#discussion_r3714501965
##########
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-40259: a scan merge preserves the sources' reported key-grouped
partitioning") {
Review Comment:
```suggestion
test("SPARK-58549: a scan merge preserves the sources' reported
key-grouped partitioning") {
```
##########
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-40259: do not merge DSv2 scans reporting incompatible
kGP/ordering") {
Review Comment:
```suggestion
test("SPARK-58549: do not merge DSv2 scans reporting incompatible
kGP/ordering") {
```
##########
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-40259: do not merge DSv2 scans reporting incompatible
kGP/ordering") {
+ // Both inputs report a partitioning/ordering, but on the different column
each reads, so no
+ // single rebuilt scan could keep both not-worse. combineRequired* returns
None (incompatible),
+ // so the merge is declined at the leaf -- before any rebuild -- with the
degradation configs
+ // off (the default). This is distinct from the single-side case above
(which rebuilds and then
+ // finds the re-derived report degraded): here the two inputs disagree
with each other up front.
+ def assertDeclines(withField: DataSourceV2ScanRelation =>
DataSourceV2ScanRelation): Unit = {
+ // withField applied to the "a" scan reports on a, applied to the "b"
scan reports on b.
+ val q = testRelation.select(
+
ScalarSubquery(withField(v2ScanReading("a")).groupBy()(sum($"a").as("sa"))),
+
ScalarSubquery(withField(v2ScanReading("b")).groupBy()(sum($"b").as("sb"))))
+ comparePlans(Optimize.execute(q.analyze), q.analyze)
+ }
+
+ assertDeclines(s => s.copy(keyGroupedPartitioning =
Some(Seq(s.output.head))))
+ assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head,
Ascending)))))
+ }
+
+ test("SPARK-40259: merge DSv2 scans reporting kGP/ordering when the
degradation config allows") {
Review Comment:
```suggestion
test("SPARK-58549: merge DSv2 scans reporting kGP/ordering when the
degradation config allows") {
```
##########
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-40259: do not merge DSv2 scans reporting incompatible
kGP/ordering") {
+ // Both inputs report a partitioning/ordering, but on the different column
each reads, so no
+ // single rebuilt scan could keep both not-worse. combineRequired* returns
None (incompatible),
+ // so the merge is declined at the leaf -- before any rebuild -- with the
degradation configs
+ // off (the default). This is distinct from the single-side case above
(which rebuilds and then
+ // finds the re-derived report degraded): here the two inputs disagree
with each other up front.
+ def assertDeclines(withField: DataSourceV2ScanRelation =>
DataSourceV2ScanRelation): Unit = {
+ // withField applied to the "a" scan reports on a, applied to the "b"
scan reports on b.
+ val q = testRelation.select(
+
ScalarSubquery(withField(v2ScanReading("a")).groupBy()(sum($"a").as("sa"))),
+
ScalarSubquery(withField(v2ScanReading("b")).groupBy()(sum($"b").as("sb"))))
+ comparePlans(Optimize.execute(q.analyze), q.analyze)
+ }
+
+ assertDeclines(s => s.copy(keyGroupedPartitioning =
Some(Seq(s.output.head))))
+ assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head,
Ascending)))))
+ }
+
+ test("SPARK-40259: merge DSv2 scans reporting kGP/ordering when the
degradation config allows") {
+ // With the matching degradation config on, a merge that would drop a
reported partitioning or
+ // ordering proceeds anyway (trading it for a single scan). The merged
scan re-derives no report
+ // from the non-reporting TestV2Scan, so the fused plan is the plain
column union.
+ val mergedScan = v2ScanReadingOn(v2Table, Seq("a", "b"))
+ val mergedSubquery = mergedScan
+ .groupBy()(sum($"a").as("sum_a"), sum($"b").as("sum_b"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("sum_b"), $"sum_b")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ def assertMerges(
+ withField: DataSourceV2ScanRelation => DataSourceV2ScanRelation,
confKey: String): Unit = {
+ val sub1 =
ScalarSubquery(withField(v2ScanReading("a")).groupBy()(sum($"a").as("sum_a")))
+ val sub2 =
ScalarSubquery(v2ScanReading("b").groupBy()(sum($"b").as("sum_b")))
+ val originalQuery = testRelation.select(sub1, sub2)
+ withSQLConf(confKey -> "true") {
+ comparePlans(Optimize.execute(originalQuery.analyze),
correctAnswer.analyze)
+ }
+ }
+
+ assertMerges(s => s.copy(keyGroupedPartitioning =
Some(Seq(s.output.head))),
+
SQLConf.MERGE_SUBPLANS_DSV2_ALLOW_KEY_GROUPED_PARTITIONING_DEGRADATION.key)
+ assertMerges(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head,
Ascending)))),
+ SQLConf.MERGE_SUBPLANS_DSV2_ALLOW_ORDERING_DEGRADATION.key)
+ }
+
+ test("SPARK-40259: enforce the required report on the deferred under-Filter
scan build") {
Review Comment:
```suggestion
test("SPARK-58549: enforce the required report on the deferred
under-Filter scan build") {
```
--
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]
