peter-toth commented on code in PR #57437:
URL: https://github.com/apache/spark/pull/57437#discussion_r3640250252
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:
##########
@@ -110,6 +110,11 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
require(filteringKeys.size == 1, "DPP Filters should only have a single
broadcasting key " +
"since there are no usage for multiple broadcasting keys at the moment.")
val indices = Seq(joinKeys.indexOf(filteringKeys.head))
+ val broadcastValueProjection = if
(conf.dynamicPartitionPruningBroadcastProjectionEnabled) {
+ ReusableBroadcastValueProjection.find(filteringKeys.head, filteringPlan,
partScan)
Review Comment:
**Finding 1.** Nearly everything `find` produces here is re-derivable from
the DPP's own `buildQuery`, which the planner already has -- so this may not
need to cross phases at all, and could be computed lazily only when it's used:
- the direct-reuse path in `PlanDynamicPruningFilters` already does
`createSparkPlan(planner, buildPlan)` and matches it via `sameResult`, and
`sourcePlan` is just a subtree of that same `buildQuery`;
- the filtering key is `buildKeys(broadcastKeyIndices.head)`.
Today `find` runs eagerly here for every candidate DPP whenever the config
is on, but its result is consumed only when direct reuse fails *and*
`onlyInBroadcast` -- a narrow case. Running
`find(buildKeys(broadcastKeyIndices.head), buildQuery, ...)` in the planner,
gated behind that fallback, drops the tag machinery entirely and does strictly
less optimizer work.
The one input not available at planning is `find`'s `excludedPlan` -- but
it's always the pruned leaf scan (`getFilterableTableScan` returns only
`LogicalRelation`/`HiveTableRelation`/`DataSourceV2ScanRelation`), used only
for the self-reference guard `!source.exists(_.sameResult(excludedPlan))`,
which doesn't affect the safe-superset and targets a scan the DPP already
identifies via `pruningKey`.
If carrying is deliberate instead, picking up @viirya's tag-vs-field
question: a `@transient` field in a *second parameter list* -- exactly how
`LogicalRDD` carries `session`/`originStats`/`originConstraints` and
`@transient stream` -- would fit better than the tag. It keeps `productArity`
at 7 (the existing 7-arg extractors and the `productArity === 7` test are
untouched, so "without changing the case-class shape" still holds), stays out
of `equals`/canonicalization (dedup unaffected, like today), stays transient
(meets the "don't serialize the source plan" requirement), and is carried
through transforms via `otherCopyArgs` -- removing the four manual
`copyBroadcastValueMetadataTo` overrides and the silent-drop-on-`.copy()` risk.
(`SubqueryAdaptiveBroadcastExec` would get the same treatment for the AQE path.)
--
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]
