You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2026/06/13 19:24:10 UTC
(spark) branch branch-4.x updated: [SPARK-54593][SQL] Enable DPP for materialized filtering sides
This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 6a75953e151a [SPARK-54593][SQL] Enable DPP for materialized filtering sides
6a75953e151a is described below
commit 6a75953e151af5996b53dd059b5b25e183a5ff5b
Author: Chao Sun <su...@apache.org>
AuthorDate: Sat Jun 13 21:23:30 2026 +0200
[SPARK-54593][SQL] Enable DPP for materialized filtering sides
### What changes were proposed in this pull request?
This PR extends dynamic partition pruning (DPP) eligibility for small, already
materialized filtering sides:
- `LocalRelation`, which represents locally available rows.
- `LogicalRDD` produced by `checkpoint()` or `localCheckpoint()`.
Checkpoint-created `LogicalRDD`s carry an explicit marker so that DPP is not
enabled for arbitrary `LogicalRDD` inputs that may require recomputing an
upstream query. This also keeps recursive CTE and `foreachBatch`-constructed
inputs outside the new eligibility rule.
This supersedes the unmerged approach in #53324 with narrower `LogicalRDD`
handling while addressing SPARK-54593.
### Why are the changes needed?
DPP currently requires a filtering predicate in the build-side logical plan.
When a small filtering side is already materialized as a `LocalRelation` or a
checkpointed `LogicalRDD`, that predicate is no longer present, so Spark misses
partition pruning opportunities.
This occurs for joins where a partition expression is matched to a small set of
keys, for example `concat_ws("||", hour, category) = hc_key`. Although the
expression is composed only from partition columns, the partitioned scan is not
dynamically pruned when the filtering side is materialized.
### Does this PR introduce _any_ user-facing change?
Yes. Queries joining a partitioned file-source table with a small
`LocalRelation` or checkpointed filtering side may now perform dynamic
partition pruning and scan fewer partitions. There is no API change.
### How was this patch tested?
- Added positive coverage for DPP using a `LocalRelation` build side with an
expression over partition columns.
- Added positive coverage for DPP using a `localCheckpoint()` build side with
the same expression form.
- Added negative coverage confirming that a non-checkpointed `LogicalRDD` does
not become DPP-eligible.
- Ran `build/sbt 'sql/testOnly org.apache.spark.sql.DynamicPartitionPruningV1SuiteAEOn org.apache.spark.sql.DynamicPartitionPruningV1SuiteAEOff'`.
- Ran `build/sbt sql/scalastyle sql/Test/scalastyle`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex
Closes #56071 from sunchao/dev/chao/codex/dpp-materialized-build-side.
Authored-by: Chao Sun <su...@apache.org>
Signed-off-by: Max Gekk <ma...@gmail.com>
(cherry picked from commit 0255283f7a4164a3706114ec7726518801071c21)
Signed-off-by: Max Gekk <ma...@gmail.com>
---
.../org/apache/spark/sql/classic/Dataset.scala | 6 +-
.../apache/spark/sql/execution/ExistingRDD.scala | 16 ++--
.../apache/spark/sql/execution/UnionLoopExec.scala | 3 +-
.../dynamicpruning/PartitionPruning.scala | 21 ++++-
.../spark/sql/DynamicPartitionPruningSuite.scala | 102 +++++++++++++++++++++
5 files changed, 136 insertions(+), 12 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala
index 6596fcbe091c..3b8712fe0f31 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala
@@ -582,7 +582,11 @@ class Dataset[T] private[sql](
}
withTypedPlan[T] {
- LogicalRDD.fromDataset(rdd = internalRdd, originDataset = this, isStreaming = isStreaming)
+ LogicalRDD.fromDataset(
+ rdd = internalRdd,
+ originDataset = this,
+ isStreaming = isStreaming,
+ fromCheckpoint = true)
}
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala
index 06085497de19..f2e87568632a 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala
@@ -105,7 +105,8 @@ case class LogicalRDD(
// originStats and originConstraints are intentionally placed to "second" parameter list,
// to prevent catalyst rules to mistakenly transform and rewrite them. Do not change this.
originStats: Option[Statistics] = None,
- originConstraints: Option[ExpressionSet] = None)
+ originConstraints: Option[ExpressionSet] = None,
+ fromCheckpoint: Boolean = false)
extends LeafNode
with StreamSourceAwareLogicalPlan
with MultiInstanceRelation {
@@ -113,7 +114,7 @@ case class LogicalRDD(
import LogicalRDD._
override protected final def otherCopyArgs: Seq[AnyRef] =
- session :: originStats :: originConstraints :: Nil
+ session :: originStats :: originConstraints :: Boolean.box(fromCheckpoint) :: Nil
override def newInstance(): LogicalRDD.this.type = {
val rewrite = Utils.toMap(output, output.map(_.newInstance()))
@@ -141,7 +142,7 @@ case class LogicalRDD(
rewrittenOrdering,
isStreaming,
stream
- )(session, rewrittenStatistics, rewrittenConstraints).asInstanceOf[this.type]
+ )(session, rewrittenStatistics, rewrittenConstraints, fromCheckpoint).asInstanceOf[this.type]
}
override protected def stringArgs: Iterator[Any] = Iterator(output, isStreaming)
@@ -156,6 +157,8 @@ case class LogicalRDD(
}
}
+ private[sql] def isCheckpointedInput: Boolean = fromCheckpoint
+
override lazy val constraints: ExpressionSet = originConstraints.getOrElse(ExpressionSet())
// Subqueries can have non-deterministic results even when they only contain deterministic
// expressions (e.g. consider a LIMIT 1 subquery without an ORDER BY). Propagating predicates
@@ -166,7 +169,7 @@ case class LogicalRDD(
.filterNot(SubqueryExpression.hasSubquery)
override def withStream(stream: SparkDataStream): LogicalRDD = {
- copy(stream = Some(stream))(session, originStats, originConstraints)
+ copy(stream = Some(stream))(session, originStats, originConstraints, fromCheckpoint)
}
override def getStream: Option[SparkDataStream] = stream
@@ -181,7 +184,8 @@ object LogicalRDD extends Logging {
private[sql] def fromDataset(
rdd: RDD[InternalRow],
originDataset: Dataset[_],
- isStreaming: Boolean): LogicalRDD = {
+ isStreaming: Boolean,
+ fromCheckpoint: Boolean = false): LogicalRDD = {
// Takes the first leaf partitioning whenever we see a `PartitioningCollection`. Otherwise the
// size of `PartitioningCollection` may grow exponentially for queries involving deep inner
// joins.
@@ -206,7 +210,7 @@ object LogicalRDD extends Logging {
executedPlan.outputOrdering,
isStreaming,
None
- )(originDataset.sparkSession, stats, constraints)
+ )(originDataset.sparkSession, stats, constraints, fromCheckpoint)
}
private[sql] def buildOutputAssocForRewrite(
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala
index c6d4091fc809..0d8fb4c814e4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/UnionLoopExec.scala
@@ -229,7 +229,8 @@ case class UnionLoopExec(
val optimizedPlan = prevDF.queryExecution.optimizedPlan
val (stats, constraints) = rewriteStatsAndConstraints(r, optimizedPlan,
sameOutput = false)
- logicalRDD.copy(output = r.output)(prevDF.sparkSession, stats, constraints)
+ logicalRDD.copy(output = r.output)(
+ prevDF.sparkSession, stats, constraints, fromCheckpoint = false)
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala
index fd3416204448..ca7c8442d5f9 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala
@@ -24,6 +24,7 @@ import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.connector.read.SupportsRuntimeV2Filtering
+import org.apache.spark.sql.execution.LogicalRDD
import org.apache.spark.sql.execution.columnar.InMemoryRelation
import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation}
import org.apache.spark.sql.execution.datasources.v2.ExtractV2Scan
@@ -199,11 +200,22 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join
/**
- * Search a filtering predicate in a given logical plan
+ * Search for a selective filtering operation, a LocalRelation, or a checkpoint-derived
+ * LogicalRDD.
+ *
+ * LocalRelation rows are already locally available. A checkpoint-derived LogicalRDD establishes
+ * an explicit checkpoint boundary and can be used as a broadcast build side for DPP without
+ * evaluating the computation upstream of that boundary again.
+ *
+ * InMemoryRelation is intentionally excluded because cache() and persist() are lazy: its
+ * presence does not guarantee the cached data has been materialized, and missing or evicted
+ * blocks may require evaluating the upstream computation again.
*/
- private def hasSelectivePredicate(plan: LogicalPlan): Boolean = {
+ private def hasSelectivePredicateOrLocalOrCheckpointedInput(plan: LogicalPlan): Boolean = {
plan.exists {
case f: Filter => isLikelySelective(f.condition)
+ case _: LocalRelation => true
+ case r: LogicalRDD => r.isCheckpointedInput
case _ => false
}
}
@@ -212,10 +224,11 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join
* To be able to prune partitions on a join key, the filtering side needs to
* meet the following requirements:
* (1) it can not be a stream
- * (2) it needs to contain a selective predicate used for filtering
+ * (2) it needs to contain a selective predicate, a LocalRelation, or a checkpoint-derived
+ * LogicalRDD
*/
private def hasPartitionPruningFilter(plan: LogicalPlan): Boolean = {
- !plan.isStreaming && hasSelectivePredicate(plan)
+ !plan.isStreaming && hasSelectivePredicateOrLocalOrCheckpointedInput(plan)
}
private def prune(plan: LogicalPlan): LogicalPlan = {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
index c68f64d52f2a..d303a03ba64b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
@@ -22,6 +22,7 @@ import org.scalatest.GivenWhenThen
import org.apache.spark.sql.catalyst.expressions.{DynamicPruningExpression, Expression}
import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode._
import org.apache.spark.sql.catalyst.plans.ExistenceJoin
+import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
import org.apache.spark.sql.connector.catalog.{InMemoryTableCatalog, InMemoryTableWithV2FilterCatalog}
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive._
@@ -1695,6 +1696,107 @@ abstract class DynamicPartitionPruningV1Suite extends DynamicPartitionPruningDat
import testImplicits._
+ test("DPP with a checkpointed filtering side and an expression partition key") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") {
+ withTable("events") {
+ Seq((1, "hour1", "a"), (2, "hour1", "b"), (3, "hour2", "a"))
+ .toDF("id", "hour", "category")
+ .write
+ .partitionBy("hour", "category")
+ .format(tableFormat)
+ .mode("overwrite")
+ .saveAsTable("events")
+
+ val sampledKeys = Seq(("hour1", "a"))
+ .toDF("hour", "category")
+ .select(concat_ws("||", $"hour", $"category").as("hc_key"))
+ .localCheckpoint(eager = true)
+
+ assert(sampledKeys.queryExecution.optimizedPlan.exists {
+ case r: LogicalRDD => r.isCheckpointedInput
+ case _ => false
+ })
+
+ val events = spark.table("events").as("events")
+ val sampled = sampledKeys.as("sampled")
+ val df = events
+ .join(broadcast(sampled),
+ concat_ws("||", $"events.hour", $"events.category") === $"sampled.hc_key")
+ .select($"events.id")
+
+ checkPartitionPruningPredicate(df, withSubquery = false, withBroadcast = true)
+ checkAnswer(df, Row(1))
+ }
+ }
+ }
+
+ test("DPP with a local filtering side and an expression partition key") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") {
+ withTable("events") {
+ Seq((1, "hour1", "a"), (2, "hour1", "b"), (3, "hour2", "a"))
+ .toDF("id", "hour", "category")
+ .write
+ .partitionBy("hour", "category")
+ .format(tableFormat)
+ .mode("overwrite")
+ .saveAsTable("events")
+
+ val sampledKeys = Seq(("hour1", "a"))
+ .toDF("hour", "category")
+ .select(concat_ws("||", $"hour", $"category").as("hc_key"))
+
+ assert(sampledKeys.queryExecution.optimizedPlan.exists(_.isInstanceOf[LocalRelation]))
+
+ val events = spark.table("events").as("events")
+ val sampled = sampledKeys.as("sampled")
+ val df = events
+ .join(broadcast(sampled),
+ concat_ws("||", $"events.hour", $"events.category") === $"sampled.hc_key")
+ .select($"events.id")
+
+ checkPartitionPruningPredicate(df, withSubquery = false, withBroadcast = true)
+ checkAnswer(df, Row(1))
+ }
+ }
+ }
+
+ test("DPP does not treat a non-checkpointed LogicalRDD as a selective filtering side") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") {
+ withTable("events") {
+ Seq((1, "hour1", "a"), (2, "hour1", "b"))
+ .toDF("id", "hour", "category")
+ .write
+ .partitionBy("hour", "category")
+ .format(tableFormat)
+ .mode("overwrite")
+ .saveAsTable("events")
+
+ val originalKeys = Seq("hour1||a").toDF("hc_key")
+ val keys: DataFrame = LogicalRDD.fromDataset(
+ rdd = originalKeys.queryExecution.toRdd,
+ originDataset = originalKeys,
+ isStreaming = false)
+ assert(keys.queryExecution.optimizedPlan.exists {
+ case r: LogicalRDD => !r.isCheckpointedInput
+ case _ => false
+ })
+
+ val events = spark.table("events").as("events")
+ val sampled = keys.as("sampled")
+ val df = events
+ .join(broadcast(sampled),
+ concat_ws("||", $"events.hour", $"events.category") === $"sampled.hc_key")
+ .select($"events.id")
+
+ checkPartitionPruningPredicate(df, withSubquery = false, withBroadcast = false)
+ checkAnswer(df, Row(1))
+ }
+ }
+ }
+
/**
* Check the static scan metrics with and without DPP
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org