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 5b5e0f6fe1aa [SPARK-54593][SQL][FOLLOWUP] Require a statistics-based
benefit, not the fallback ratio, for materialized-input DPP
5b5e0f6fe1aa is described below
commit 5b5e0f6fe1aa488d2de170b18c41c82918373e5f
Author: Wenchen Fan <[email protected]>
AuthorDate: Mon Jun 29 09:34:27 2026 +0800
[SPARK-54593][SQL][FOLLOWUP] Require a statistics-based benefit, not the
fallback ratio, for materialized-input DPP
### What changes were proposed in this pull request?
Follow-up to https://github.com/apache/spark/pull/56071 (SPARK-54593),
which made an already-materialized filtering side (a `LocalRelation` or a
`checkpoint()` / `localCheckpoint()`-derived `LogicalRDD`) eligible for dynamic
partition pruning (DPP), and to https://github.com/apache/spark/pull/56636,
which re-framed that structural eligibility as a recompute-cost guard
(`isCheaplyRecomputableMaterializedPlan`).
Such a side has no selective predicate, so it should obtain DPP only
through **broadcast reuse** or a **genuine statistics-based benefit** -- not
through `pruningHasBenefit`'s no-statistics fallback ratio, which is meant for
"stats missing, but there is a predicate likely to be selective". This PR:
- Gates the fallback ratio on `hasSelectivePredicate`. A materialized-only
side with no usable statistics reports no benefit (broadcast-reuse only)
instead of being injected as a standalone, always-applied subquery on a guessed
ratio.
- Always honors a statistics-based ratio. When the per-column statistic is
absent, a materialized side's exact `maxRows` is used as a conservative upper
bound on the join-key NDV, so a small, selective materialized side still
establishes a real benefit and keeps standalone DPP.
This builds on the structural eligibility predicates already on master
(`hasSelectivePredicate` / `isCheaplyRecomputableMaterializedPlan`) and changes
only the benefit estimate in `pruningHasBenefit`.
### Why are the changes needed?
With `spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly =
false`, a materialized side that prunes little or nothing (no selective
predicate, no usable statistics) was injected as an always-applied subquery
that re-executes the filtering side for no benefit. Before SPARK-54593 such a
side was not DPP-eligible at all, so this is a regression in master. The
broadcast-reuse path (the feature's intended use) and genuinely beneficial
standalone DPP (proven by statistics, includ [...]
### Does this PR introduce _any_ user-facing change?
No. It only avoids planning a no-benefit dynamic partition pruning subquery
for a materialized build side; query results are unchanged.
### How was this patch tested?
New unit test in `DynamicPartitionPruningSuite` ("DPP materialized-input
eligibility requires an estimable pruning benefit"), with `reuseBroadcastOnly =
false` and no reusable broadcast:
- a small `LocalRelation` (no column statistics) keeps standalone DPP via
its exact `maxRows` bound;
- a checkpoint-derived `LogicalRDD` with no retained statistics and no
`maxRows` is **not** injected as a standalone subquery;
- an opaque (`mapPartitions`-derived) side is not eligible, with or without
a sibling broadcast.
Existing positive tests for materialized filtering sides (broadcast reuse)
continue to pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic Claude Opus)
Closes #56603 from cloud-fan/SPARK-54593-followup.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 5 +-
.../dynamicpruning/PartitionPruning.scala | 83 ++++++---
.../spark/sql/DynamicPartitionPruningSuite.scala | 194 ++++++++++++++++++---
3 files changed, 227 insertions(+), 55 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index ff2dd2dbd483..2d4d9a75c374 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -700,7 +700,10 @@ object SQLConf {
.doc("When statistics are not available or configured not to be used,
this config will be " +
"used as the fallback filter ratio for computing the data size of the
partitioned table " +
"after dynamic partition pruning, in order to evaluate if it is worth
adding an extra " +
- "subquery as the pruning filter if broadcast reuse is not applicable.")
+ "subquery as the pruning filter if broadcast reuse is not applicable.
This fallback " +
+ "ratio only applies when the filtering side has a selective predicate;
a materialized " +
+ "filtering side without one obtains dynamic partition pruning through
broadcast reuse " +
+ "only.")
.version("3.0.0")
.doubleConf
.createWithDefault(0.5)
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 25de90c1c15f..3e1b39b564c7 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
@@ -111,7 +111,7 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
"since there are no usage for multiple broadcasting keys at the moment.")
val indices = Seq(joinKeys.indexOf(filteringKeys.head))
lazy val hasBenefit = pruningHasBenefit(
- pruningKey, partScan, filteringKeys.head, filteringPlan)
+ pruningKey, partScan, filteringKeys.head, filteringPlan,
hasSelectivePredicate(filteringPlan))
if (reuseEnabled || hasBenefit) {
// insert a DynamicPruning wrapper to identify the subquery during query
planning
Filter(
@@ -134,45 +134,74 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
* in bytes of the plan on the other side of the join. We estimate the
filtering ratio
* using column statistics if they are available, otherwise we use the
config value of
* `spark.sql.optimizer.dynamicPartitionPruning.fallbackFilterRatio`.
+ *
+ * The fallback ratio is only meaningful "when CBO stats are missing, but
there is a predicate
+ * that is likely to be selective" -- so it is used only when
`hasSelectivePredicate` is true. A
+ * filtering side that is eligible only because it is already materialized
(a LocalRelation or a
+ * checkpoint-derived LogicalRDD, SPARK-54593) carries no such predicate;
for it we rely solely on
+ * the statistics-based ratio and report no benefit when statistics are
unavailable, so it is not
+ * injected as a standalone always-applied subquery on a guessed ratio. A
statistics-based ratio,
+ * when available, is always honored regardless of `hasSelectivePredicate`.
*/
private def pruningHasBenefit(
partExpr: Expression,
partPlan: LogicalPlan,
otherExpr: Expression,
- otherPlan: LogicalPlan): Boolean = {
+ otherPlan: LogicalPlan,
+ hasSelectivePredicate: Boolean): Boolean = {
// get the distinct counts of an attribute for a given table
def distinctCounts(attr: Attribute, plan: LogicalPlan): Option[BigInt] = {
plan.stats.attributeStats.get(attr).flatMap(_.distinctCount)
}
- // the default filtering ratio when CBO stats are missing, but there is a
- // predicate that is likely to be selective
- val fallbackRatio = conf.dynamicPartitionPruningFallbackFilterRatio
- // the filtering ratio based on the type of the join condition and on the
column statistics
- val filterRatio = (partExpr.references.toList,
otherExpr.references.toList) match {
- // filter out expressions with more than one attribute on any side of
the operator
- case (leftAttr :: Nil, rightAttr :: Nil)
- if conf.dynamicPartitionPruningUseStats =>
- // get the CBO stats for each attribute in the join condition
- val partDistinctCount = distinctCounts(leftAttr, partPlan)
- val otherDistinctCount = distinctCounts(rightAttr, otherPlan)
- val availableStats = partDistinctCount.isDefined &&
partDistinctCount.get > 0 &&
- otherDistinctCount.isDefined
- if (!availableStats) {
- fallbackRatio
- } else if (partDistinctCount.get.toDouble <=
otherDistinctCount.get.toDouble) {
- // there is likely an estimation error, so we fallback
- fallbackRatio
- } else {
- 1 - otherDistinctCount.get.toDouble /
partDistinctCount.get.toDouble
- }
- case _ => fallbackRatio
+ // the filtering ratio derived from column statistics, when reliable stats
are available
+ val statsBasedRatio: Option[Double] =
+ (partExpr.references.toList, otherExpr.references.toList) match {
+ // filter out expressions with more than one attribute on any side of
the operator
+ case (leftAttr :: Nil, rightAttr :: Nil)
+ if conf.dynamicPartitionPruningUseStats =>
+ // get the CBO stats for each attribute in the join condition
+ val partDistinctCount = distinctCounts(leftAttr, partPlan)
+ // A materialized filtering side (e.g. a LocalRelation) may carry
no column statistics
+ // but an exact `maxRows`, which is a conservative upper bound on
its join-key NDV. Use
+ // it when the column statistic is missing so a small, selective
materialized side still
+ // yields a statistics-based ratio rather than falling through to
the gated fallback.
+ // Restrict this to a cheaply-recomputable materialized side: DPP
re-evaluates the
+ // filtering side, so deriving a pruning benefit from the
`maxRows` of an opaque plan
+ // (e.g. one with a `mapPartitions`, which
`isCheaplyRecomputableMaterializedPlan`
+ // rejects) could inject a standalone subquery that prunes a
partition the join's
+ // re-evaluation then needs, giving wrong results for a
hidden-non-deterministic side.
+ val otherDistinctCount =
+ distinctCounts(rightAttr, otherPlan).orElse {
+ if (isCheaplyRecomputableMaterializedPlan(otherPlan)) {
+ otherPlan.maxRows.map(BigInt(_))
+ } else {
+ None
+ }
+ }
+ val availableStats = partDistinctCount.isDefined &&
partDistinctCount.get > 0 &&
+ otherDistinctCount.isDefined
+ if (!availableStats) {
+ None
+ } else if (partDistinctCount.get.toDouble <=
otherDistinctCount.get.toDouble) {
+ // there is likely an estimation error, so there is no reliable
stats-based ratio
+ None
+ } else {
+ Some(1 - otherDistinctCount.get.toDouble /
partDistinctCount.get.toDouble)
+ }
+ case _ => None
+ }
+
+ // Without a reliable stats-based ratio, fall back to the configured ratio
only when there is a
+ // predicate likely to be selective; otherwise there is no evidence of a
pruning benefit.
+ val filterRatio = statsBasedRatio.orElse {
+ if (hasSelectivePredicate)
Some(conf.dynamicPartitionPruningFallbackFilterRatio) else None
}
- val estimatePruningSideSize = filterRatio *
partPlan.stats.sizeInBytes.toFloat
- val overhead = calculatePlanOverhead(otherPlan)
- estimatePruningSideSize > overhead
+ filterRatio.exists { ratio =>
+ ratio * partPlan.stats.sizeInBytes.toFloat >
calculatePlanOverhead(otherPlan)
+ }
}
/**
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 de44286593b9..4aa4a8870b77 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
@@ -1766,6 +1766,108 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
}
}
+ test("SPARK-54593: a materialized filtering side is not injected as a
standalone DPP subquery " +
+ "without an estimated pruning benefit") {
+ // Broadcast joins are disabled, so the DPP filter cannot reuse a
broadcast exchange. A
+ // materialized filtering side carries no selective predicate, so its
pruning benefit cannot be
+ // estimated; it must therefore not be injected as an always-applied
standalone subquery (which
+ // would re-execute the filtering side and prune nothing for a
non-selective side).
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ 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")
+
+ // A materialized filtering side that covers every partition: it
offers no pruning benefit.
+ val sampledKeys = Seq("hour1||a", "hour1||b",
"hour2||a").toDF("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(sampled, concat_ws("||", $"events.hour", $"events.category")
=== $"sampled.hc_key")
+ .select($"events.id")
+
+ checkPartitionPruningPredicate(df, withSubquery = false, withBroadcast
= false)
+ checkAnswer(df, Row(1) :: Row(2) :: Row(3) :: Nil)
+ }
+ }
+ }
+
+ test("SPARK-54593: a materialized filtering side keeps statistics-backed
standalone DPP") {
+ // A checkpoint-derived LogicalRDD has no Filter but can retain column
statistics. When those
+ // statistics establish a pruning benefit, DPP must still be injected as a
standalone subquery
+ // when no broadcast can be reused -- the materialized-input handling
gates only the
+ // no-statistics fallback ratio, not the statistics-based benefit.
+ withSQLConf(
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_USE_STATS.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ // The fallback ratio offers no benefit, so DPP can only come from the
statistics-based ratio.
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "0",
+ // No broadcast can be reused, so an injected filter must be a
standalone subquery.
+ SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false",
+ SQLConf.CBO_ENABLED.key -> "true") {
+ withTable("dim_one") {
+ // A dimension with a single distinct store_id, with column statistics
computed.
+ spark.range(20).selectExpr("1 AS store_id")
+ .write.format(tableFormat).saveAsTable("dim_one")
+ sql("ANALYZE TABLE dim_one COMPUTE STATISTICS FOR COLUMNS store_id")
+
+ // Checkpoint a projection of it: a LogicalRDD with no Filter that
retains the NDV (= 1).
+ val keys =
spark.table("dim_one").select("store_id").localCheckpoint(eager = true)
+ val keysPlan = keys.queryExecution.optimizedPlan
+ assert(keysPlan.exists {
+ case r: LogicalRDD => r.isCheckpointedInput
+ case _ => false
+ })
+
assert(keysPlan.stats.attributeStats.values.exists(_.distinctCount.contains(BigInt(1))),
+ s"checkpointed side should retain the NDV statistic:\n$keysPlan")
+
+ // fact_stats is partitioned by store_id (NDV > 1) and has column
statistics, so the
+ // statistics-based ratio establishes a pruning benefit for this
materialized side.
+ val df = spark.table("fact_stats").join(keys,
"store_id").select("date_id")
+
+ checkPartitionPruningPredicate(df, withSubquery = true, withBroadcast
= false)
+ }
+ }
+ }
+
+ test("SPARK-54593: a small materialized filtering side keeps standalone DPP
via its row bound") {
+ // A LocalRelation has no column statistics but an exact maxRows, a
conservative upper bound on
+ // its join-key NDV. A small, selective materialized side must still get
standalone DPP when no
+ // broadcast can be reused -- the row bound, not the gated fallback ratio,
supplies the benefit.
+ withSQLConf(
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_USE_STATS.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ // The fallback ratio offers no benefit, so DPP can only come from the
row-bound ratio.
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "0",
+ // No broadcast can be reused, so an injected filter must be a
standalone subquery.
+ SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false",
+ SQLConf.CBO_ENABLED.key -> "true") {
+ // A one-row LocalRelation: no column statistics, but maxRows = 1 bounds
the join-key NDV.
+ val keys = Seq(0).toDF("store_id")
+ val keysPlan = keys.queryExecution.optimizedPlan
+ assert(keysPlan.exists(_.isInstanceOf[LocalRelation]))
+ assert(keysPlan.maxRows.contains(1))
+
assert(!keysPlan.stats.attributeStats.values.exists(_.distinctCount.isDefined),
+ s"LocalRelation should not carry a distinct-count
statistic:\n$keysPlan")
+
+ // fact_stats is partitioned by store_id (NDV > 1) and has column
statistics, so the row bound
+ // (NDV <= 1) establishes a strong pruning benefit for this materialized
side.
+ val df = spark.table("fact_stats").join(keys,
"store_id").select("date_id")
+
+ checkPartitionPruningPredicate(df, withSubquery = true, withBroadcast =
false)
+ }
+ }
+
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") {
@@ -1840,11 +1942,17 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
}
}
- test("DPP materialized-input eligibility requires a cheaply recomputable
plan") {
+ test("DPP materialized-input eligibility requires an estimable pruning
benefit") {
+ // A materialized filtering side is injected as a standalone DPP subquery
only when it has an
+ // estimable pruning benefit. With CBO statistics on the partitioned table
(the partition-side
+ // NDV) and a small row bound on the filtering side (a LocalRelation's
exact maxRows), the
+ // benefit comes from the statistics-based ratio. The fallback ratio is no
longer applied to a
+ // side without a selective predicate, so it is pinned to 0 to keep it out
of the picture.
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",
- SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1000",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_USE_STATS.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "0",
+ SQLConf.CBO_ENABLED.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false",
SQLConf.SUBQUERY_REUSE_ENABLED.key -> "false",
@@ -1859,6 +1967,10 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
.mode("overwrite")
.saveAsTable("events")
+ // Column statistics on the partitioned table supply the
partition-side NDV used to estimate
+ // the pruning benefit for a materialized filtering side.
+ sql("ANALYZE TABLE events COMPUTE STATISTICS FOR COLUMNS p")
+
def activeDppSubqueries(df: DataFrame): Seq[InSubqueryExec] = {
collectDynamicPruningExpressions(df.queryExecution.executedPlan)
.collect { case in: InSubqueryExec => in }
@@ -1871,7 +1983,7 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
assert(activeDppSubqueries(df).exists {
case InSubqueryExec(_, _: SubqueryExec, _, _, _, _) => true
case _ => false
- }, s"Should execute standalone DPP for a cheaply recomputable
materialized plan:\n" +
+ }, s"Should execute standalone DPP for a materialized plan with an
estimable benefit:\n" +
df.queryExecution)
}
@@ -1880,33 +1992,20 @@ 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 an opaque materialized plan:\n" +
+ s"Shouldn't execute standalone DPP for this filtering side:\n" +
df.queryExecution)
}
+ // A LocalRelation is cheaply recomputable and exposes an exact
maxRows, a conservative
+ // bound on its join-key NDV, so its pruning benefit is estimable and
it gets a standalone
+ // DPP subquery.
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")))
+ // A checkpoint-derived LogicalRDD is also cheaply recomputable, but a
checkpointed
+ // LocalRelation retains no column statistics and exposes no maxRows,
so its pruning benefit
+ // cannot be estimated; with no broadcast to reuse it is not injected
as a DPP subquery at
+ // all. The statistics-backed checkpointed case (retained NDV) is
covered by a dedicated
+ // test.
+ checkNoDpp(Seq(1).toDF("p").localCheckpoint(eager = true))
val checkpointed = Seq(1).toDS().localCheckpoint(eager = true)
val mappedKeys = checkpointed.mapPartitions { values =>
@@ -1915,6 +2014,14 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
}.toDF("p")
checkNoDpp(mappedKeys)
+ // An opaque side becomes DPP-eligible through a selective predicate,
and a `limit` gives it
+ // an exact maxRows. That maxRows must not be a pruning-benefit NDV
bound for a side that is
+ // not cheaply recomputable: a standalone subquery would re-evaluate
the stateful
+ // mapPartitions, prune the fact to the first key, and drop the rows
the join's second
+ // evaluation needs. With the maxRows bound gated on
isCheaplyRecomputableMaterializedPlan,
+ // no benefit is estimated, no standalone DPP is injected, and the
result stays correct.
+ checkNoDpp(mappedKeys.filter($"p" > 0).limit(1))
+
withSQLConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true") {
val broadcastJoin =
spark.table("events").join(broadcast(mappedKeys),
Seq("p")).select("p")
@@ -1924,6 +2031,39 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
s"Shouldn't trigger DPP for an opaque broadcast plan:\n" +
broadcastJoin.queryExecution)
+ // The cheaply-recomputable structural branches of
isCheaplyRecomputableMaterializedPlan
+ // are exercised through broadcast reuse: a checkpointed input
carries no statistics and
+ // no maxRows, so it has no estimable benefit and only the
eligibility decision (not the
+ // benefit term) determines whether the broadcast is reused for DPP.
+ val cpKeys = Seq(1).toDF("p").localCheckpoint(eager = true)
+
+ def checkBroadcastReuseEligible(side: DataFrame): Unit = {
+ checkPartitionPruningPredicate(
+ spark.table("events").join(broadcast(side),
Seq("p")).select("p"),
+ withSubquery = false, withBroadcast = true)
+ }
+
+ def checkBroadcastReuseIneligible(side: DataFrame): Unit = {
+ checkPartitionPruningPredicate(
+ spark.table("events").join(broadcast(side),
Seq("p")).select("p"),
+ withSubquery = false, withBroadcast = false)
+ }
+
+ // A non-selective Filter/Project above a materialized input stays
eligible: its recompute
+ // cost is dominated by the materialized leaf, so it reuses the
broadcast for DPP even
+ // though the boolean-cast filter is not classified selective by
isLikelySelective.
+
checkBroadcastReuseEligible(cpKeys.filter($"p".cast("boolean")).select($"p"))
+
+ // An Aggregate over a materialized input is excluded: its
shuffle/compute cost is
+ // invisible to the scan-bytes cost model, so it is not DPP-eligible
even with a reusable
+ // broadcast.
+
checkBroadcastReuseIneligible(cpKeys.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.
+ val identityUdf = udf((x: Int) => x)
+
checkBroadcastReuseIneligible(cpKeys.select(identityUdf($"p").as("p")))
+
val target = spark.table("events").hint("merge")
.join(mappedKeys.hint("merge"), Seq("p"))
.select($"p", lit("target").as("branch"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
