sunchao commented on code in PR #56603:
URL: https://github.com/apache/spark/pull/56603#discussion_r3455503845
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:
##########
@@ -134,45 +134,63 @@ 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.
+ val otherDistinctCount =
+ distinctCounts(rightAttr,
otherPlan).orElse(otherPlan.maxRows.map(BigInt(_)))
Review Comment:
[P1] I rechecked this after the #56636 rebase at exact head `7ed155cb`
against current base `acb7de23`, and the focused A/B test still reproduces the
regression:
- base: no standalone DPP, one evaluation, 1000 expected rows;
- head: standalone DPP, two evaluations, 0 rows.
The #56636 cleanup does not catch this case: the stateful
`MapPartitions.func` is not a Catalyst expression, while
`QueryPlan.deterministic` inspects node expressions and children, so the DPP
predicate survives cleanup. I agree that the broader hidden-nondeterminism
limitation predates this PR when a positive fallback already enables DPP; the
PR-introduced regression is specifically the supported `fallbackFilterRatio=0`
configuration, where the base is correct and this unrestricted `maxRows` path
turns on duplicated evaluation.
Could we restrict the `maxRows` fallback to
`isCheaplyRecomputableMaterializedPlan(otherPlan)`? That preserves the intended
LocalRelation/checkpoint cases without attempting to solve repeatability
generally.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:
##########
@@ -134,45 +134,63 @@ 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 {
Review Comment:
[P2] I rechecked this on exact current base `acb7de23` and head `7ed155cb`
after the rebase. The focused A/B case still gives:
- base: standalone DPP, 1/20 partitions scanned;
- head: no DPP, 20/20 partitions scanned.
The exact-reference limitation predates this PR, but the demonstrated
regression does not: gating the fallback exposes the miss even though the new
row-bound evidence is valid. Since `maxRows` bounds the NDV of any expression
over those rows, could its fallback be applied independently of
`otherExpr.references.size`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
