You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pt...@apache.org on 2026/03/20 12:05:52 UTC
(spark) branch master updated: [SPARK-55848][SQL][TESTS] Add regression tests for SPJ partial clustering dedup
This is an automated email from the ASF dual-hosted git repository.
ptoth pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new c67c02a26b09 [SPARK-55848][SQL][TESTS] Add regression tests for SPJ partial clustering dedup
c67c02a26b09 is described below
commit c67c02a26b093f2c660e6dd98614a69efaf96f97
Author: Naveen Kumar Puppala <na...@gmail.com>
AuthorDate: Fri Mar 20 13:05:22 2026 +0100
[SPARK-55848][SQL][TESTS] Add regression tests for SPJ partial clustering dedup
### 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.
Closes #54714 from naveenp2708/spark-55848-tests-master.
Authored-by: Naveen Kumar Puppala <na...@gmail.com>
Signed-off-by: Peter Toth <pe...@gmail.com>
---
.../connector/KeyGroupedPartitioningSuite.scala | 141 ++++++++++++++++++++-
1 file changed, 140 insertions(+), 1 deletion(-)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
index c29be45b72fa..4116de1b8978 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
@@ -29,7 +29,7 @@ import org.apache.spark.sql.connector.catalog.functions._
import org.apache.spark.sql.connector.distributions.Distributions
import org.apache.spark.sql.connector.expressions._
import org.apache.spark.sql.connector.expressions.Expressions._
-import org.apache.spark.sql.execution.{ExtendedMode, FormattedMode, SimpleMode, SparkPlan}
+import org.apache.spark.sql.execution.{ExtendedMode, FormattedMode, RDDScanExec, SimpleMode, SparkPlan}
import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2ScanRelation, GroupPartitionsExec}
import org.apache.spark.sql.execution.exchange.{ShuffleExchangeExec, ShuffleExchangeLike}
import org.apache.spark.sql.execution.joins.SortMergeJoinExec
@@ -1213,6 +1213,145 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with
}
}
+ test("SPARK-55848: dropDuplicates after SPJ with partial clustering") {
+ 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"""
+ |${selectWithMergeJoinHint("i", "p")} 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") {
+ 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 (
+ | ${selectWithMergeJoinHint("i", "p")} 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)))
+
+ // One GroupPartitionsExec per join child to align the partially-clustered
+ // partitions, and one above the join to group for the window.
+ 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 window)")
+ }
+ }
+
+ test("SPARK-55848: checkpointed partially-clustered join with 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.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> true.toString) {
+ // Checkpoint the JOIN result (not the scan) so the checkpoint node carries the
+ // partially-clustered KeyGroupedPartitioning. The dedup on top must still insert
+ // the required grouping operator because partially-clustered partitioning does not
+ // satisfy ClusteredDistribution.
+ val joinedDf = sql(
+ s"""${selectWithMergeJoinHint("i", "p")} i.id, i.name, i.price
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id = p.item_id""".stripMargin)
+ val checkpointedDf = joinedDf.checkpoint()
+ val df = checkpointedDf.select("id").distinct()
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
+
+ val checkpointScans = collect(df.queryExecution.executedPlan) {
+ case r: RDDScanExec => r
+ }
+ assert(checkpointScans.exists(_.outputPartitioning match {
+ case kp: physical.KeyedPartitioning => !kp.isGrouped
+ case _ => false
+ }), "checkpoint (RDDScanExec) should have ungrouped KeyedPartitioning")
+
+ val allGroupPartitions = collectAllGroupPartitions(df.queryExecution.executedPlan)
+ assert(allGroupPartitions.size === 1,
+ "expected 1 GroupPartitionsExec above the checkpointed join for dedup")
+ }
+ }
+ }
+
test("data source partitioning + dynamic partition filtering") {
withSQLConf(
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org