You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "naveenp2708 (via GitHub)" <gi...@apache.org> on 2026/03/10 05:07:01 UTC
[PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
naveenp2708 opened a new pull request, #54714:
URL: https://github.com/apache/spark/pull/54714
### What changes were proposed in this pull request?
Test-only PR. Adds regression tests for SPARK-55848 (SPJ partial clustering produces incorrect results for post-join dedup operations).
Three tests added to KeyGroupedPartitioningSuite:
1. SPARK-55848: dropDuplicates after SPJ with partial clustering
2. SPARK-55848: Window dedup after SPJ with partial clustering
3. SPARK-55848: checkpointed scan with partial clustering and dedup
### Why are the changes needed?
The fix was merged via #54330, but regression tests for the correctness issue (SPARK-55848 / #54378) were not included. These tests ensure the issue does not regress.
### Does this PR introduce any user-facing change?
No. Test-only change.
### How was this patch tested?
All 73 tests in KeyGroupedPartitioningSuite pass.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL][TESTS] Add regression tests for SPJ partial clustering dedup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4097567899
Thank you @naveenp2708.
Merged to `master` (4.2.0).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54714:
URL: https://github.com/apache/spark/pull/54714#discussion_r2912588045
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,115 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
Review Comment:
Can you please check if we have the expected `GroupPartitionExec`s in the plan? I believe one should be under the join to align partitions to the other, partially clustered side of the join; and one should be above the join to group partitions for the aggregate.
Also please assert that we don't have exchanges in the plan.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4038891573
@naveenp2708, can you please take a look at https://github.com/apache/spark/pull/54751#discussion_r2918016286 and adjust the test case there and here as well? Actually, first probably we should agree on the fix in https://github.com/apache/spark/pull/54751 and then update this PR.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54714:
URL: https://github.com/apache/spark/pull/54714#discussion_r2912588045
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,115 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
Review Comment:
Can you please check if we have `GroupPartitionExec`s in the plan at the expected points? I believe one should be under the join to align partitions to the other, partially clustered side of the join; and one should be above the join to group partitions for the aggregate.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "naveenp2708 (via GitHub)" <gi...@apache.org>.
naveenp2708 commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4032965697
@peter-toth Added GroupPartitionsExec assertions to all 3 tests:
- 2 GroupPartitionsExec under the join (one per child to align partitions)
- 1 GroupPartitionsExec above the join (for aggregate/window grouping)
- Total of 3 GroupPartitionsExec per test p
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "naveenp2708 (via GitHub)" <gi...@apache.org>.
naveenp2708 commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4035154674
@peter-toth Enabled the workflows on my fork — "Build" and "Report test results" both passed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "naveenp2708 (via GitHub)" <gi...@apache.org>.
naveenp2708 commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4028725168
@peter-toth Per your suggestion — tests-only PR for master. Branch-4.1 PR with fix + tests coming next.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #54714:
URL: https://github.com/apache/spark/pull/54714#discussion_r2928048296
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,142 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
+
+ // One GroupPartitionsExec per join child to align the partially-clustered
+ // partitions, and one above the join to group for the aggregate.
+ val joinGP = collectGroupPartitions(df.queryExecution.executedPlan)
+ assert(joinGP.size === 2,
+ "expected 2 GroupPartitionsExec under the join")
+ val allGP = collectAllGroupPartitions(df.queryExecution.executedPlan)
+ assert(allGP.size === 3,
+ "expected 3 GroupPartitionsExec total (2 under join + 1 above for aggregate)")
+ }
+ }
+
+ test("SPARK-55848: Window dedup after SPJ with partial clustering should produce " +
Review Comment:
nit: 'should produce correct results' is redundant :)
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,142 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
+
+ // One GroupPartitionsExec per join child to align the partially-clustered
+ // partitions, and one above the join to group for the aggregate.
+ val joinGP = collectGroupPartitions(df.queryExecution.executedPlan)
+ assert(joinGP.size === 2,
+ "expected 2 GroupPartitionsExec under the join")
+ val allGP = collectAllGroupPartitions(df.queryExecution.executedPlan)
+ assert(allGP.size === 3,
+ "expected 3 GroupPartitionsExec total (2 under join + 1 above for aggregate)")
+ }
+ }
+
+ test("SPARK-55848: Window dedup after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // Use ROW_NUMBER() OVER to dedup joined rows per id after a partially-clustered
+ // SPJ. The WINDOW operator requires ClusteredDistribution on i.id; with partial
+ // clustering the plan must insert the right exchange/group so that the window
+ // produces exactly one row per id.
+ val df = sql(
Review Comment:
should we use 'selectWithMergeJoinHint' for consistency?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL][TESTS] Add regression tests for SPJ partial clustering dedup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth closed pull request #54714: [SPARK-55848][SQL][TESTS] Add regression tests for SPJ partial clustering dedup
URL: https://github.com/apache/spark/pull/54714
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54714:
URL: https://github.com/apache/spark/pull/54714#discussion_r2912567588
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,115 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
Review Comment:
Can you please check if we have the expected shuffle in the plan here and other tests as well?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4029690300
Seems like check didn't run on this PR yet, most likely because you haven't enabled Github Actions in your forked repo.
Can you please check the "Pull request" section at https://spark.apache.org/contributing.html? Or if you click on the [pending check](https://github.com/apache/spark/pull/54714/checks?check_run_id=66405136772) that also shows you how to enable/retrigger the necessary workflows.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54714:
URL: https://github.com/apache/spark/pull/54714#discussion_r2912567588
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,115 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
Review Comment:
Can you please check if we have the expected shuffle in the plan here and other tests as well?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "naveenp2708 (via GitHub)" <gi...@apache.org>.
naveenp2708 commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4051930855
@szehon-ho Fixed both — shortened test names and switched to selectWithMergeJoinHint. Thanks for the review!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL][TESTS] Add regression tests for SPJ partial clustering dedup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4084323098
> @peter-toth Fixed the title to include [TESTS] tag. Retriggered the checks.
Thank you @naveenp2708. Could you please retrigger the tests again?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL][TESTS] Add regression tests for SPJ partial clustering dedup [spark]
Posted by "naveenp2708 (via GitHub)" <gi...@apache.org>.
naveenp2708 commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4084424148
@peter-toth Retriggered. Thanks for the approval!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL][TESTS] Add regression tests for SPJ partial clustering dedup [spark]
Posted by "naveenp2708 (via GitHub)" <gi...@apache.org>.
naveenp2708 commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4095446062
@peter-toth @dongjoon-hyun Resolved the merge conflict and retriggered CI.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54714:
URL: https://github.com/apache/spark/pull/54714#discussion_r2912671678
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,115 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
+ }
+ }
+
+ test("SPARK-55848: Window dedup after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // Use ROW_NUMBER() OVER to dedup joined rows per id after a partially-clustered
+ // SPJ. The WINDOW operator requires ClusteredDistribution on i.id; with partial
+ // clustering the plan must insert the right exchange/group so that the window
+ // produces exactly one row per id.
+ val df = sql(
+ s"""
+ |SELECT id, price FROM (
+ | SELECT i.id, i.price,
+ | ROW_NUMBER() OVER (PARTITION BY i.id ORDER BY i.price DESC) AS rn
+ | FROM testcat.ns.$items i
+ | JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |) t WHERE rn = 1
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1, 41.0f), Row(2, 10.0f), Row(3, 15.5f)))
+ }
+ }
+
+ test("SPARK-55848: checkpointed scan with partial clustering and dedup") {
+ withTempDir { dir =>
+ spark.sparkContext.setCheckpointDir(dir.getPath)
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // Checkpoint the items scan, then join with purchases under partial clustering,
+ // and finally dedup. The checkpointed scan preserves the KeyedPartitioning
+ // with non-grouped keys; the dedup must still produce correct distinct ids.
+ val itemsDf = spark.read.table(s"testcat.ns.$items").checkpoint()
Review Comment:
I mean in this test we should checkpoint the partially clustered result of a join and put an aggregate on top of that checkpoint. The aggregate should require a `GroupPartitionsExec` on `master` or a shuffle on previous branches (before the refactor).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54714:
URL: https://github.com/apache/spark/pull/54714#discussion_r2912671678
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,115 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
+ }
+ }
+
+ test("SPARK-55848: Window dedup after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // Use ROW_NUMBER() OVER to dedup joined rows per id after a partially-clustered
+ // SPJ. The WINDOW operator requires ClusteredDistribution on i.id; with partial
+ // clustering the plan must insert the right exchange/group so that the window
+ // produces exactly one row per id.
+ val df = sql(
+ s"""
+ |SELECT id, price FROM (
+ | SELECT i.id, i.price,
+ | ROW_NUMBER() OVER (PARTITION BY i.id ORDER BY i.price DESC) AS rn
+ | FROM testcat.ns.$items i
+ | JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |) t WHERE rn = 1
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1, 41.0f), Row(2, 10.0f), Row(3, 15.5f)))
+ }
+ }
+
+ test("SPARK-55848: checkpointed scan with partial clustering and dedup") {
+ withTempDir { dir =>
+ spark.sparkContext.setCheckpointDir(dir.getPath)
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // Checkpoint the items scan, then join with purchases under partial clustering,
+ // and finally dedup. The checkpointed scan preserves the KeyedPartitioning
+ // with non-grouped keys; the dedup must still produce correct distinct ids.
+ val itemsDf = spark.read.table(s"testcat.ns.$items").checkpoint()
Review Comment:
I mean i this test we should checkpoint the result of a join with partially clustering and put an aggregate on top of that checkpoint. The aggregate should require a `GroupPartitionsExec` on `master` or a shuffle on previous branches (before the refactor).
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,115 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
+ }
+ }
+
+ test("SPARK-55848: Window dedup after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // Use ROW_NUMBER() OVER to dedup joined rows per id after a partially-clustered
+ // SPJ. The WINDOW operator requires ClusteredDistribution on i.id; with partial
+ // clustering the plan must insert the right exchange/group so that the window
+ // produces exactly one row per id.
+ val df = sql(
+ s"""
+ |SELECT id, price FROM (
+ | SELECT i.id, i.price,
+ | ROW_NUMBER() OVER (PARTITION BY i.id ORDER BY i.price DESC) AS rn
+ | FROM testcat.ns.$items i
+ | JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |) t WHERE rn = 1
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1, 41.0f), Row(2, 10.0f), Row(3, 15.5f)))
+ }
+ }
+
+ test("SPARK-55848: checkpointed scan with partial clustering and dedup") {
+ withTempDir { dir =>
+ spark.sparkContext.setCheckpointDir(dir.getPath)
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // Checkpoint the items scan, then join with purchases under partial clustering,
+ // and finally dedup. The checkpointed scan preserves the KeyedPartitioning
+ // with non-grouped keys; the dedup must still produce correct distinct ids.
+ val itemsDf = spark.read.table(s"testcat.ns.$items").checkpoint()
Review Comment:
I mean in this test we should checkpoint the result of a join with partially clustering and put an aggregate on top of that checkpoint. The aggregate should require a `GroupPartitionsExec` on `master` or a shuffle on previous branches (before the refactor).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "naveenp2708 (via GitHub)" <gi...@apache.org>.
naveenp2708 commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4058237226
@peter-toth Makes sense — PR #54751 now has approvals from you and @szehon-ho. Once it's merged, I'll update this PR to align the tests. Already addressed szehon-ho's nits (shortened test names + selectWithMergeJoinHint).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #54714:
URL: https://github.com/apache/spark/pull/54714#discussion_r2912588045
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1210,6 +1210,115 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering should produce " +
+ "correct results") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ // Two rows for id=1 so partial clustering may split them across tasks
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // dropDuplicates on the join key after a partially-clustered SPJ must still
+ // produce the correct number of distinct ids. Before the fix, the
+ // partially-clustered partitioning was incorrectly treated as satisfying
+ // ClusteredDistribution, so EnsureRequirements did not insert an Exchange
+ // before the dedup, leading to duplicate rows.
+ val df = sql(
+ s"""
+ |SELECT DISTINCT i.id
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
Review Comment:
Can you please check if we have the expected GroupPartitionExecs in the plan? I believe one should be under the join to align partitions to the other, partially clustered side of the join; and one should be above the join to group partitions for the aggregate.
Also please assert that we don't have any exchanges in the plan.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "naveenp2708 (via GitHub)" <gi...@apache.org>.
naveenp2708 commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4072160437
@peter-toth Updated title with [TESTS] tag. Tests aligned with merged #54751 — checkpointed test checkpoints join result, AQE disabled, GroupPartitionsExec assertions, shortened names, selectWithMergeJoinHint. Will open backport PRs for branch-4.0 and branch-3.5 next.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4073993325
@naveenp2708 , can you please double check the PR title? It looks incomplete and I don't see `[TESTS]` there; and also retrigger the checks (the failure seems unrelated)?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55848][SQL] Add regression tests for SPJ partial clustering de… [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #54714:
URL: https://github.com/apache/spark/pull/54714#issuecomment-4069602346
@naveenp2708 , I merged https://github.com/apache/spark/pull/54751, can you please update this PR? Also, let's add `[TESTS]` tag to this one.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org