You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/09/07 02:20:10 UTC
(spark) branch master updated: [SPARK-59199][SQL] SPJ: Support runtime partition filtering for more join types
This is an automated email from the ASF dual-hosted git repository.
pan3793 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 64eff668530f [SPARK-59199][SQL] SPJ: Support runtime partition filtering for more join types
64eff668530f is described below
commit 64eff668530f18aa23de229da051e5dba755b5ce
Author: Cheng Pan <pa...@gmail.com>
AuthorDate: Mon Sep 7 10:19:58 2026 +0800
[SPARK-59199][SQL] SPJ: Support runtime partition filtering for more join types
### What changes were proposed in this pull request?
Extend the per-join-type partition filtering that `spark.sql.sources.v2.bucketing.partition.filter.enabled` applies in `EnsureRequirements.mergeAndDedupPartitions` to the join types that were falling through to the no-filtering default:
| Key groups kept | Join types |
|---|---|
| present on both sides | `Inner`, **`Cross`**, **`LeftSemi`** |
| left side's | `LeftOuter`, **`LeftAnti`**, **`LeftSingle`**, **`ExistenceJoin`** |
| right side's | `RightOuter` |
| both sides' (no filtering) | `FullOuter`, any other |
Bold entries are new. Only equi-joins reach this path (every sort-merge and shuffled-hash join takes its keys from `ExtractEquiJoinKeys`), so a `Cross` join here always carries equi keys and filters like `Inner`.
### Why are the changes needed?
In a storage-partitioned join, rows only match within a key group. A key group that no output row can come from is wasted scan work:
- `Cross` and `LeftSemi` emit nothing from a group missing on either side.
- `LeftAnti`, `LeftSingle` and `ExistenceJoin` keep or test every left row, so a right-only group contributes nothing.
Those join types previously kept the union of both sides' key groups, i.e. no filtering.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `EnsureRequirementsSuite`: new unit test asserting the merged key groups for all nine join types, with the config on and off.
- `KeyGroupedPartitioningSuite`:
- new end-to-end tests for `CROSS ... ON`, `LEFT SEMI`, `LEFT ANTI`, an existence join (`EXISTS ... OR`) and a left single join (correlated scalar subquery), each asserting the planned join type, no shuffle, the query result and the number of retained key groups;
- a condition-less `CROSS JOIN` test asserting it does not reach SPJ and keeps every pair;
- the SPARK-59176 test extended with a leg emptied by a semi join, so a reduced side with no key still joins.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Fable 5.1)
Closes #58514 from pan3793/spj-partition-filter-join-types.
Authored-by: Cheng Pan <pa...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../execution/exchange/EnsureRequirements.scala | 12 +-
.../connector/KeyGroupedPartitioningSuite.scala | 158 +++++++++++++++++++--
.../exchange/EnsureRequirementsSuite.scala | 36 ++++-
3 files changed, 188 insertions(+), 18 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 2a33280861da..493f816f8003 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
@@ -851,11 +851,19 @@ case class EnsureRequirements(
joinType: JoinType,
keyOrdering: Ordering[InternalRowComparableWrapper]): Seq[InternalRowComparableWrapper] = {
val merged = if (SQLConf.get.getConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED)) {
+ // Rows with matching join keys land in the same key group. If a group is absent from one
+ // side, whether it can produce output depends on which side's unmatched rows the join
+ // preserves. Only equi-joins reach this method, since every SMJ/SHJ takes its keys from
+ // `ExtractEquiJoinKeys`. So a Cross join, e.g. `l CROSS JOIN r ON l.id = r.id`, is treated
+ // like an Inner join: groups absent from either side cannot produce output.
joinType match {
- case Inner =>
+ // neither side keeps unmatched rows
+ case _: InnerLike | LeftSemi =>
mergeAndDedupPartitionKeys(leftPartitionKeys, rightPartitionKeys, intersect = true)
- case LeftOuter => leftPartitionKeys.distinct
+ // every left row is kept or tested
+ case LeftOuter | LeftAnti | LeftSingle | ExistenceJoin(_) => leftPartitionKeys.distinct
case RightOuter => rightPartitionKeys.distinct
+ // FullOuter keeps both sides' unmatched rows; any other join type is not filtered
case _ => mergeAndDedupPartitionKeys(leftPartitionKeys, rightPartitionKeys)
}
} else {
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 6a911ce65261..7a16e8e6a781 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
@@ -24,6 +24,7 @@ import org.apache.spark.rdd.SortedMergeCoalescedRDD
import org.apache.spark.sql.{DataFrame, ExplainSuiteHelper, Row}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Ascending, AttributeReference, ExprId, Literal, TransformExpression}
+import org.apache.spark.sql.catalyst.plans.{Cross, ExistenceJoin, JoinType, LeftAnti, LeftSemi, LeftSingle}
import org.apache.spark.sql.catalyst.plans.physical
import org.apache.spark.sql.connector.catalog.{Column, Identifier, InMemoryCatalystRuntimeFilterCatalog, InMemoryTableCatalog}
import org.apache.spark.sql.connector.catalog.functions._
@@ -42,7 +43,7 @@ import org.apache.spark.sql.execution.{
UnionExec}
import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2ScanRelation, GroupPartitionsExec}
import org.apache.spark.sql.execution.exchange.{ShuffleExchangeExec, ShuffleExchangeLike, ValidateRequirements}
-import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, ShuffledJoin, SortMergeJoinExec}
import org.apache.spark.sql.functions.{col, max}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf._
@@ -564,6 +565,49 @@ class KeyGroupedPartitioningSuite
|""".stripMargin)
}
+ /**
+ * Creates `items` with ids 0, 1, 4 and `purchases` with item ids 4, 5, both bucketed by those
+ * columns, so a join on them sees two left-only key groups, one shared and one right-only, then
+ * runs `body` with partition filtering enabled.
+ */
+ private def withPartitionFilterJoinTables(body: => Unit): Unit = {
+ createTable(items, itemsColumns, Array(bucket(8, "id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ s"(0, 'aa', 38.0, cast('2020-01-01' as timestamp)), " +
+ s"(1, 'bb', 39.0, cast('2020-01-02' as timestamp)), " +
+ s"(4, 'cc', 40.0, cast('2020-01-02' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array(bucket(8, "item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ s"(4, 42.0, cast('2020-01-01' as timestamp)), " +
+ s"(5, 44.0, cast('2020-01-15' as timestamp))")
+
+ withSQLConf(SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "true") {
+ body
+ }
+ }
+
+ /**
+ * Asserts `df` plans exactly one shuffled join accepted by `isJoinType`, that no shuffle was
+ * added under it, and that partition filtering left it `expectedNumGroups` key groups.
+ */
+ private def checkPartitionFilteredJoin(
+ df: DataFrame,
+ isJoinType: JoinType => Boolean,
+ expectedNumGroups: Int): Unit = {
+ val plan = df.queryExecution.executedPlan
+ val joins = collect(plan) { case j: ShuffledJoin if isJoinType(j.joinType) => j }
+ assert(joins.size == 1, s"expected one matching join in\n$plan")
+ assert(collectAllShuffles(joins.head).isEmpty,
+ "should not add shuffle for both sides of the join")
+ val groupPartitions = collectAllGroupPartitions(joins.head)
+ assert(groupPartitions.nonEmpty, s"expected GroupPartitionsExec in\n$plan")
+ val actualNumGroups = groupPartitions.map(_.outputPartitioning.numPartitions)
+ assert(actualNumGroups.forall(_ == expectedNumGroups),
+ s"expected $expectedNumGroups key groups, got ${actualNumGroups.mkString(", ")}")
+ }
+
/**
* Creates a table partitioned by `bucket(numBuckets, id)` and holding the ids 0 until `numIds`.
* Joining two such tables reduces both sides onto the greatest common divisor of their bucket
@@ -632,12 +676,20 @@ class KeyGroupedPartitioningSuite
/**
* Joins `days1` to `years1` and `days2` to `years2`, each reducing both of its sides onto the
* year key space, then joins the two reduced legs to each other with `joinType`. `leg2First` puts
- * the second leg on the left of that join. The projection takes the timestamp from whichever side
- * has it, so an outer join reports the same rows in either order.
+ * the second leg on the left of that join. `leg2Semi` makes the second leg a semi join, with
+ * `years2` on its left so that the leg still projects that side. The projection takes the
+ * timestamp from whichever side has it, so an outer join reports the same rows in either order.
*/
- private def reducedTsLegJoin(leg2First: Boolean = false, joinType: String = "JOIN"): String = {
+ private def reducedTsLegJoin(
+ leg2First: Boolean = false,
+ leg2Semi: Boolean = false,
+ joinType: String = "JOIN"): String = {
val leg1 = "SELECT d.ts FROM testcat.ns.days1 d JOIN testcat.ns.years1 y ON y.ts = d.ts"
- val leg2 = "SELECT y.ts FROM testcat.ns.days2 d JOIN testcat.ns.years2 y ON y.ts = d.ts"
+ val leg2 = if (leg2Semi) {
+ "SELECT y.ts FROM testcat.ns.years2 y LEFT SEMI JOIN testcat.ns.days2 d ON y.ts = d.ts"
+ } else {
+ "SELECT y.ts FROM testcat.ns.days2 d JOIN testcat.ns.years2 y ON y.ts = d.ts"
+ }
val (left, right) = if (leg2First) (leg2, leg1) else (leg1, leg2)
s"SELECT coalesce(l.ts, r.ts) AS ts FROM ($left) l $joinType ($right) r ON l.ts = r.ts"
}
@@ -1134,16 +1186,18 @@ class KeyGroupedPartitioningSuite
// Both orders, since the side that has no key is the one to leave out of the comparison.
// And both join types, since the inner join intersects the two key sets to nothing and so
// has nothing to sort, while the full outer join keeps the other side's keys and sorts them
- // by the reported types.
- Seq("JOIN" -> Nil, "FULL OUTER JOIN" -> bothTimestamps).foreach {
- case (joinType, expected) =>
- Seq(false, true).foreach { leg2First =>
- val df = sql(reducedTsLegJoin(leg2First, joinType))
-
- checkAnswer(df, expected)
- assert(collectShuffles(stripAQEPlan(df.queryExecution.executedPlan)).isEmpty,
- "the two legs are the same pairing, so all three joins are co-partitioned")
- }
+ // by the reported types. And a semi join emptying the second leg, since SPARK-59199 makes
+ // it intersect like the inner join.
+ for {
+ (joinType, expected) <- Seq("JOIN" -> Nil, "FULL OUTER JOIN" -> bothTimestamps)
+ leg2First <- Seq(false, true)
+ leg2Semi <- Seq(false, true)
+ } {
+ val df = sql(reducedTsLegJoin(leg2First, leg2Semi, joinType))
+
+ checkAnswer(df, expected)
+ assert(collectShuffles(stripAQEPlan(df.queryExecution.executedPlan)).isEmpty,
+ "the two legs are the same pairing, so all three joins are co-partitioned")
}
}
}
@@ -3573,6 +3627,80 @@ class KeyGroupedPartitioningSuite
}
}
+ // A cross join reaches SPJ only with equi keys, so it and a semi join drop the key groups absent
+ // on either side. An anti join has to probe every left row, so only the right-only groups are
+ // dropped.
+ Seq(
+ ("CROSS", Cross, Seq(Row(4, "cc", 40.0)), 1),
+ ("LEFT SEMI", LeftSemi, Seq(Row(4, "cc", 40.0)), 1),
+ ("LEFT ANTI", LeftAnti, Seq(Row(0, "aa", 38.0), Row(1, "bb", 39.0)), 3)
+ ).foreach { case (joinSql, joinType, expectedRows, expectedNumGroups) =>
+ test(s"SPARK-59199: test partition filters with $joinSql join") {
+ withPartitionFilterJoinTables {
+ val df = sql(
+ s"""
+ |${selectWithMergeJoinHint("i", "p")}
+ |id, name, i.price
+ |FROM testcat.ns.$items i $joinSql JOIN testcat.ns.$purchases p
+ |ON i.id = p.item_id
+ |ORDER BY id
+ |""".stripMargin)
+ checkAnswer(df, expectedRows)
+ checkPartitionFilteredJoin(df, _ == joinType, expectedNumGroups)
+ }
+ }
+ }
+
+ test("SPARK-59199: a cross join without an equi condition does not reach SPJ") {
+ withPartitionFilterJoinTables {
+ val df = sql(
+ s"""
+ |SELECT i.id, p.item_id
+ |FROM testcat.ns.$items i CROSS JOIN testcat.ns.$purchases p
+ |""".stripMargin)
+ val plan = df.queryExecution.executedPlan
+ assert(collectAllGroupPartitions(plan).isEmpty,
+ s"a cartesian product must not group partitions in\n$plan")
+ assert(df.count() == 6, "every left x right pair survives")
+ }
+ }
+
+ test("SPARK-59199: test partition filters with existence join") {
+ withPartitionFilterJoinTables {
+ // EXISTS in a disjunction is planned as an ExistenceJoin
+ val df = sql(
+ s"""
+ |SELECT id, name, price FROM testcat.ns.$items i
+ |WHERE EXISTS (SELECT /*+ MERGE(p) */ 1 FROM testcat.ns.$purchases p
+ | WHERE i.id = p.item_id)
+ | OR i.name = 'bb'
+ |ORDER BY id
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(1, "bb", 39.0), Row(4, "cc", 40.0)))
+ // an existence join tests every left row, so only the right-only key groups are dropped
+ checkPartitionFilteredJoin(df, _.isInstanceOf[ExistenceJoin], expectedNumGroups = 3)
+ }
+ }
+
+ test("SPARK-59199: test partition filters with left single join") {
+ withPartitionFilterJoinTables {
+ // a correlated scalar subquery not proven to return one row is planned as a LeftSingle
+ // join, which cannot sort-merge, hence the shuffled hash join hint
+ withSQLConf(SQLConf.SCALAR_SUBQUERY_USE_SINGLE_JOIN.key -> "true") {
+ val df = sql(
+ s"""
+ |SELECT id, name,
+ | (SELECT /*+ SHUFFLE_HASH(p) */ p.price FROM testcat.ns.$purchases p
+ | WHERE i.id = p.item_id) AS sale_price
+ |FROM testcat.ns.$items i
+ |""".stripMargin)
+ checkAnswer(df, Seq(Row(0, "aa", null), Row(1, "bb", null), Row(4, "cc", 42.0)))
+ // a left single join keeps every left row, so only the right-only key groups are dropped
+ checkPartitionFilteredJoin(df, _ == LeftSingle, expectedNumGroups = 3)
+ }
+ }
+ }
+
test("SPARK-53322: checkpointed scans avoid shuffles for aggregates") {
withTempDir { dir =>
spark.sparkContext.setCheckpointDir(dir.getPath)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala
index 1d877f169605..c5e368624d53 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala
@@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.DirectShufflePartitionID
import org.apache.spark.sql.catalyst.expressions.aggregate.Sum
import org.apache.spark.sql.catalyst.optimizer.BuildRight
-import org.apache.spark.sql.catalyst.plans.Inner
+import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.physical.{SinglePartition, _}
import org.apache.spark.sql.catalyst.statsEstimation.StatsTestPlan
import org.apache.spark.sql.connector.catalog.functions._
@@ -1169,6 +1169,40 @@ class EnsureRequirementsSuite extends SharedSparkSession {
}
}
+ test("SPARK-59199: mergeAndDedupPartitions filters partitions per join type") {
+ def partitioning(values: Int*): KeyedPartitioning =
+ KeyedPartitioning(Seq(exprA), values.map(v => InternalRow(v)))
+ val left = partitioning(1, 2, 3)
+ val right = partitioning(2, 3, 4)
+ val intersected = partitioning(2, 3).partitionKeys
+ val union = partitioning(1, 2, 3, 4).partitionKeys
+ def merge(joinType: JoinType): Seq[InternalRow] =
+ EnsureRequirements.mergeAndDedupPartitions(
+ left.partitionKeys, right.partitionKeys, joinType, left.keyOrdering).map(_.row)
+
+ val expected = Seq(
+ Inner -> intersected,
+ Cross -> intersected,
+ LeftSemi -> intersected,
+ LeftOuter -> left.partitionKeys,
+ LeftAnti -> left.partitionKeys,
+ LeftSingle -> left.partitionKeys,
+ ExistenceJoin(exprA) -> left.partitionKeys,
+ RightOuter -> right.partitionKeys,
+ FullOuter -> union)
+
+ withSQLConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "true") {
+ expected.foreach { case (joinType, keys) =>
+ assert(merge(joinType) === keys.map(_.row), joinType)
+ }
+ }
+ withSQLConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "false") {
+ expected.foreach { case (joinType, _) =>
+ assert(merge(joinType) === union.map(_.row), joinType)
+ }
+ }
+ }
+
test("SPARK-42168: FlatMapCoGroupInPandas and Window function with differing key order") {
val lKey = AttributeReference("key", IntegerType)()
val lKey2 = AttributeReference("key2", IntegerType)()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org