This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 25a949f18572 [SPARK-54593][SQL][FOLLOWUP] Frame materialized-input DPP
eligibility as a recompute-cost guard
25a949f18572 is described below
commit 25a949f1857214b06cbaacc29a8090d7e97a3469
Author: Wenchen Fan <[email protected]>
AuthorDate: Mon Jun 22 11:41:29 2026 -0700
[SPARK-54593][SQL][FOLLOWUP] Frame materialized-input DPP eligibility as a
recompute-cost guard
### What changes were proposed in this pull request?
Follow-up to #56535 (SPARK-54593). That PR let a filtering side without a
selective predicate become DPP-eligible when it is built from an
already-materialized input (a `LocalRelation`, or a checkpoint-derived
`LogicalRDD`), restricting the operators above the materialized leaf to a
narrow set (`Project`/`Filter`/`Union`/`SubqueryAlias`, additionally requiring
expressions to be deterministic and free of subqueries/UDFs). That restriction
was documented as ensuring the plan is "repeatable".
This PR re-frames the check (`isRepeatableMaterializedPlan` ->
`isCheaplyRecomputableMaterializedPlan`) around the reason the restriction
actually exists -- **recompute cost** -- and drops the parts that were really a
piecemeal repeatability guard:
- The operator allowlist is kept and justified on cost grounds. A subquery
in a `Project`/`Filter` is still excluded, re-justified as cost (it embeds its
own plan, whose recompute cost `calculatePlanOverhead` cannot see).
- The expression-level determinism / UDF / generator check is **removed**
(see below).
- The tests from #56535 are kept, and two cases are added that pin the
operators-above cost behavior.
### Why are the changes needed?
"Repeatability" was the wrong justification. DPP evaluates the filtering
side independently from the join (a standalone subquery, or a reused broadcast
versus a shuffled join input), so it has *always* assumed the filtering side is
repeatable, on every eligibility path including the selective-predicate one.
Materialization does not add a repeatability guarantee, and the allowlist did
not establish one in general.
What materialization actually provides is the **cost** counterpart to a
selective predicate. A selective predicate is evidence of a high pruning ratio
(the benefit term of `pruningHasBenefit`); an already-materialized input makes
the side ~free to re-read (the cost term, `calculatePlanOverhead`), so even a
modest pruning ratio is worthwhile. That cost claim only holds when the
operators above the materialized leaf are dominated by their input's scan bytes
-- exactly what `calculatePla [...]
Repeatability is deliberately **not** checked here. Whether re-evaluating
the filtering side yields the same rows is a pre-existing, DPP-wide concern
(the selective-predicate path carries it too -- a non-deterministic operator
over any filtering side can produce different keys on re-evaluation), and it
should be addressed by a system-level design rather than patched piecemeal on
the materialized path. Accordingly the expression-level determinism / UDF check
is removed; a non-determini [...]
### Does this PR introduce _any_ user-facing change?
No. DPP is an optimization; results are unchanged for repeatable filtering
sides, which is the supported case.
### How was this patch tested?
Existing `DynamicPartitionPruning*Suite`s, including the two tests from
#56535. Added two cases to the materialized-input eligibility test: a cheap
`filter`/`select` chain over a checkpoint stays eligible (standalone DPP), and
an `Aggregate` over a checkpoint is excluded (its shuffle cost is invisible to
the cost model).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic Claude Opus)
Closes #56636 from cloud-fan/dpp-materialized-input-followup.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/sql/execution/SparkOptimizer.scala | 8 ++-
.../dynamicpruning/PartitionPruning.scala | 79 +++++++++++++++-------
.../spark/sql/DynamicPartitionPruningSuite.scala | 31 +++++++--
3 files changed, 87 insertions(+), 31 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala
index e27faf7b4d9e..1b3b2d3efc72 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala
@@ -112,7 +112,13 @@ class SparkOptimizer(
V2ScanRelationPushDown.ruleName,
V2ScanPartitioningAndOrdering.ruleName,
V2Writes.ruleName,
- ReplaceCTERefWithRepartition.ruleName)
+ ReplaceCTERefWithRepartition.ruleName,
+ // CleanupDynamicPruningFilters finalizes the DPP predicates inserted by
PartitionPruning --
+ // notably rewriting non-deterministic ones to `true` so they are not
re-evaluated. That is
+ // correctness behavior, not an optional optimization, so the rule must
not be excludable.
+ // Disabling DPP is done by excluding PartitionPruning (the inserter),
after which this rule
+ // is a no-op.
+ CleanupDynamicPruningFilters.ruleName)
/**
* Optimization batches that are executed before the regular optimization
batches (also before
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 93e388c45af0..25de90c1c15f 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
@@ -205,38 +205,66 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
}
/**
- * Returns whether a plan can be evaluated repeatedly from materialized
inputs and produce the
- * same rows.
+ * Returns whether the filtering side is cheap enough to recompute that DPP
is worthwhile even
+ * without a selective predicate: its cost is dominated by an
already-materialized input, with
+ * only scan-cost-bound operators above it.
*
- * 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.
+ * This is the cost-side counterpart to `hasSelectivePredicate`. A selective
predicate is
+ * evidence of a high pruning ratio (the benefit term of
`pruningHasBenefit`); an
+ * already-materialized input is the complementary signal on the cost term
-- a `LocalRelation`
+ * (rows already local) or a checkpoint-derived `LogicalRDD`
(`isCheckpointedInput` requires the
+ * RDD to be actually checkpointed, so a lazy checkpoint does not qualify)
is ~free to re-read,
+ * so even a modest pruning ratio clears the benefit bar. `InMemoryRelation`
is excluded because
+ * cache()/persist() are lazy: its presence does not guarantee the data has
been materialized,
+ * and missing or evicted blocks may require recomputing the upstream plan.
*
- * 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.
+ * The operators above the materialized input are restricted to ones whose
cost is dominated by
+ * their input's scan bytes -- the only cost `calculatePlanOverhead` can
see. `Project`/`Filter`
+ * add negligible compute, a `Union`'s cost is the sum of its (materialized)
children, and
+ * `SubqueryAlias` is a no-op. `Aggregate`, joins, and opaque RDD operators
(e.g. `mapPartitions`)
+ * are excluded: they add compute or a shuffle the scan-bytes cost model
cannot see, so treating
+ * such a side as a cheap materialized input would overstate the pruning
benefit. A `Project`/
+ * `Filter` is likewise excluded when its expressions embed a subquery
(which carries its own
+ * plan) or an opaque user function (a UDF or a user-defined generator) --
both add recompute
+ * cost `calculatePlanOverhead` does not account for.
*
- * The supported operators are intentionally narrow. DPP is optional, and
logical-plan
- * determinism does not cover user functions stored outside Catalyst
expressions.
+ * This is primarily a cost guard, but the eligible shapes are also
repeatable in practice, which
+ * matters because DPP duplicates the filtering side and must produce the
same keys on
+ * re-evaluation. Honest non-determinism does not slip through: a `rand()`
(or a UDF marked
+ * non-deterministic) above the materialized input makes the resulting
`DynamicPruningSubquery`
+ * non-deterministic (`PlanExpression.deterministic` folds in its build
plan), so
+ * `CleanupDynamicPruningFilters` rewrites the dynamic predicate to `true`
before physical
+ * planning rather than planning a standalone `SubqueryExec` -- it is never
re-evaluated. That
+ * rule is non-excludable (`SparkOptimizer.nonExcludableRules`), so this
holds regardless of
+ * `spark.sql.optimizer.excludedRules`. The
+ * residual, DPP-wide limitation is *hidden* non-determinism left marked
deterministic; the
+ * opaque-expression exclusion above narrows it, and the rest is
intentionally left to a future
+ * system-level design rather than patched piecemeal here. The one
materialized-input-specific
+ * repeatability concern -- a checkpoint that has not been materialized yet
-- is handled by
+ * `LogicalRDD.isCheckpointedInput` requiring the RDD to be actually
checkpointed.
*/
- private def isRepeatableMaterializedPlan(plan: LogicalPlan): Boolean = {
- def isRepeatableExpression(expression: Expression): Boolean = {
- expression.deterministic && !SubqueryExpression.hasSubquery(expression)
&&
- !expression.exists {
- case _: NonSQLExpression | _: UserDefinedExpression | _:
UserDefinedGenerator => true
- case _ => false
- }
+ private def isCheaplyRecomputableMaterializedPlan(plan: LogicalPlan):
Boolean = {
+ // An expression keeps the side cheap only if its cost is bounded by the
input scan that
+ // `calculatePlanOverhead` measures. A subquery embeds its own plan, and
an opaque user
+ // function (a Scala/Python UDF, a user-defined generator, or any other
non-Catalyst
+ // expression) adds CPU/IO the scan-bytes cost model cannot see --
recomputing either would
+ // cost more than the materialized leaf suggests, so it disqualifies the
side.
+ def isScanCostBoundExpression(e: Expression): Boolean = {
+ !SubqueryExpression.hasSubquery(e) && !e.exists {
+ case _: NonSQLExpression | _: UserDefinedExpression | _:
UserDefinedGenerator => true
+ case _ => false
+ }
}
plan match {
case _: LocalRelation => true
case r: LogicalRDD => r.isCheckpointedInput
- case Project(projectList, child) if
projectList.forall(isRepeatableExpression) =>
- isRepeatableMaterializedPlan(child)
- case Filter(condition, child) if isRepeatableExpression(condition) =>
- isRepeatableMaterializedPlan(child)
- case u: Union => u.children.forall(isRepeatableMaterializedPlan)
- case SubqueryAlias(_, child) => isRepeatableMaterializedPlan(child)
+ case Project(projectList, child) if
projectList.forall(isScanCostBoundExpression) =>
+ isCheaplyRecomputableMaterializedPlan(child)
+ case Filter(condition, child) if isScanCostBoundExpression(condition) =>
+ isCheaplyRecomputableMaterializedPlan(child)
+ case u: Union => u.children.forall(isCheaplyRecomputableMaterializedPlan)
+ case SubqueryAlias(_, child) =>
isCheaplyRecomputableMaterializedPlan(child)
case _ => false
}
}
@@ -245,10 +273,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 or have a repeatable
materialized input
+ * (2) it needs to contain a selective predicate or a cheaply-recomputable
materialized input
*/
private def hasPartitionPruningFilter(plan: LogicalPlan): Boolean = {
- !plan.isStreaming && (hasSelectivePredicate(plan) ||
isRepeatableMaterializedPlan(plan))
+ !plan.isStreaming &&
+ (hasSelectivePredicate(plan) ||
isCheaplyRecomputableMaterializedPlan(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 4db67ec77479..de44286593b9 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
@@ -1840,7 +1840,7 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
}
}
- test("DPP materialized-input eligibility requires a repeatable plan") {
+ test("DPP materialized-input eligibility requires a cheaply recomputable
plan") {
withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
SQLConf.DYNAMIC_PARTITION_PRUNING_USE_STATS.key -> "false",
@@ -1871,7 +1871,7 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
assert(activeDppSubqueries(df).exists {
case InSubqueryExec(_, _: SubqueryExec, _, _, _, _) => true
case _ => false
- }, s"Should execute standalone DPP for a repeatable materialized
plan:\n" +
+ }, s"Should execute standalone DPP for a cheaply recomputable
materialized plan:\n" +
df.queryExecution)
}
@@ -1880,13 +1880,34 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
DppMaterializedInputTestState.reset(counterId)
assert(df.collect().toSeq === Seq(Row(1)))
assert(activeDppSubqueries(df).isEmpty,
- s"Shouldn't trigger DPP for a non-repeatable materialized plan:\n"
+
+ s"Shouldn't trigger DPP for an opaque materialized plan:\n" +
df.queryExecution)
}
checkStandaloneDpp(Seq(1).toDF("p"))
checkStandaloneDpp(Seq(1).toDF("p").localCheckpoint(eager = true))
+ // Cheap, scan-cost-bound operators above a materialized input stay
eligible: their
+ // recompute cost is dominated by the materialized leaf that
calculatePlanOverhead sees.
+ // The filter uses a non-selective predicate (a boolean cast is not
classified selective by
+ // isLikelySelective), so eligibility must come from
isCheaplyRecomputableMaterializedPlan
+ // rather than the selective-predicate path.
+ checkStandaloneDpp(Seq(1).toDF("p").localCheckpoint(eager = true)
+ .filter($"p".cast("boolean")).select($"p"))
+
+ // An Aggregate over a materialized input is excluded even though it
is repeatable: its
+ // shuffle/compute cost is invisible to the scan-bytes cost model, so
treating it as a cheap
+ // materialized input would overstate the pruning benefit.
+ checkNoDpp(
+ Seq(1).toDF("p").localCheckpoint(eager =
true).groupBy("p").count().select("p"))
+
+ // A UDF projection over a materialized input is excluded: an opaque
user function adds
+ // CPU/IO the scan-bytes cost model cannot see, so it would be
recomputed without that cost
+ // being counted against the pruning benefit.
+ val identityUdf = udf((x: Int) => x)
+ checkNoDpp(
+ Seq(1).toDF("p").localCheckpoint(eager =
true).select(identityUdf($"p").as("p")))
+
val checkpointed = Seq(1).toDS().localCheckpoint(eager = true)
val mappedKeys = checkpointed.mapPartitions { values =>
val key = DppMaterializedInputTestState.next(counterId)
@@ -1900,7 +1921,7 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
DppMaterializedInputTestState.reset(counterId)
assert(broadcastJoin.collect().toSeq === Seq(Row(1)))
assert(activeDppSubqueries(broadcastJoin).isEmpty,
- s"Shouldn't trigger DPP for a non-repeatable broadcast plan:\n" +
+ s"Shouldn't trigger DPP for an opaque broadcast plan:\n" +
broadcastJoin.queryExecution)
val target = spark.table("events").hint("merge")
@@ -1916,7 +1937,7 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
assert(rows.size === 1)
assert(rows.head.getString(1) === "target")
assert(activeDppSubqueries(withSiblingBroadcast).isEmpty,
- s"A sibling broadcast shouldn't make a non-repeatable plan
eligible for DPP:\n" +
+ s"A sibling broadcast shouldn't make an opaque plan eligible for
DPP:\n" +
withSiblingBroadcast.queryExecution)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
