You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2024/08/13 05:47:12 UTC
(spark) branch branch-3.4 updated: Revert "[SPARK-49179][SQL] Fix v2 multi bucketed inner joins throw AssertionError"
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new 9852e33fc819 Revert "[SPARK-49179][SQL] Fix v2 multi bucketed inner joins throw AssertionError"
9852e33fc819 is described below
commit 9852e33fc81914a4734094aa68a33f0532b40976
Author: Dongjoon Hyun <dh...@apple.com>
AuthorDate: Mon Aug 12 22:46:57 2024 -0700
Revert "[SPARK-49179][SQL] Fix v2 multi bucketed inner joins throw AssertionError"
This reverts commit 1dd558ca8a86712c9c03bf62a0bf80011c71174f.
---
.../execution/exchange/EnsureRequirements.scala | 7 ++++---
.../connector/KeyGroupedPartitioningSuite.scala | 22 ----------------------
2 files changed, 4 insertions(+), 25 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
index 2b3675d0a9bf..bc90a869fd9b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
@@ -547,7 +547,7 @@ case class EnsureRequirements(
private def createKeyGroupedShuffleSpec(
partitioning: Partitioning,
distribution: ClusteredDistribution): Option[KeyGroupedShuffleSpec] = {
- def tryCreate(partitioning: KeyGroupedPartitioning): Option[KeyGroupedShuffleSpec] = {
+ def check(partitioning: KeyGroupedPartitioning): Option[KeyGroupedShuffleSpec] = {
val attributes = partitioning.expressions.flatMap(_.collectLeaves())
val clustering = distribution.clustering
@@ -567,10 +567,11 @@ case class EnsureRequirements(
}
partitioning match {
- case p: KeyGroupedPartitioning => tryCreate(p)
+ case p: KeyGroupedPartitioning => check(p)
case PartitioningCollection(partitionings) =>
val specs = partitionings.map(p => createKeyGroupedShuffleSpec(p, distribution))
- specs.filter(_.isDefined).map(_.get).headOption
+ assert(specs.forall(_.isEmpty) || specs.forall(_.isDefined))
+ specs.head
case _ => None
}
}
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 06485f1b4ce4..cf76f6ca32ca 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
@@ -328,28 +328,6 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase {
.add("price", FloatType)
.add("time", TimestampType)
- test("SPARK-49179: Fix v2 multi bucketed inner joins throw AssertionError") {
- val cols = Array(
- Column.create("id", LongType),
- Column.create("name", StringType))
- val buckets = Array(bucket(8, "id"))
-
- withTable("t1", "t2", "t3") {
- Seq("t1", "t2", "t3").foreach { t =>
- createTable(t, cols, buckets)
- sql(s"INSERT INTO testcat.ns.$t VALUES (1, 'aa'), (2, 'bb'), (3, 'cc')")
- }
- val df = sql(
- """
- |SELECT t1.id, t2.id, t3.name FROM testcat.ns.t1
- |JOIN testcat.ns.t2 ON t1.id = t2.id
- |JOIN testcat.ns.t3 ON t1.id = t3.id
- |""".stripMargin)
- checkAnswer(df, Seq(Row(1, 1, "aa"), Row(2, 2, "bb"), Row(3, 3, "cc")))
- assert(collectShuffles(df.queryExecution.executedPlan).isEmpty)
- }
- }
-
test("partitioned join: join with two partition keys and matching & sorted partitions") {
val items_partitions = Array(bucket(8, "id"), days("arrive_time"))
createTable(items, items_schema, items_partitions)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org