peter-toth commented on code in PR #57437:
URL: https://github.com/apache/spark/pull/57437#discussion_r3644862763
##########
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:
Thanks @sunchao -- one clarification first: I'd call it "lazy derivation"
rather than "re-derivation." The idea isn't to run `find` twice; it's to not
run it in `PartitionPruning` at all, and instead run it once in
`PlanDynamicPruningFilters` / `PlanAdaptiveDynamicPruningFilters`, gated behind
the `directBroadcast.isEmpty && onlyInBroadcast` fallback. Both rules already
seem to have what `find` needs: the `SparkPlan` is built (so the
reusable-`BroadcastHashJoinExec` search the direct path already does is right
there), and the `DynamicPruningSubquery` still carries `buildQuery` (the AQE
node carries it as `buildPlan`), so `find(buildKeys(broadcastKeyIndices.head),
buildQuery, ...)` runs with no new inputs -- with one exception.
That exception is the pruned-leaf exclusion (`excludedPlan`), which I
flagged above as the one input not in the planner -- I think it's also what you
mean by carrying "the exact pruned-leaf exclusion" consistently. It's always
the pruned leaf scan and the DPP already identifies its pruning target through
`pruningKey`, so it seems re-expressible in the planner -- but if it turns out
it can't be done cleanly there, I'd take that as a fair argument for keeping
the projection carried as it is now.
The reason I'd single lazy derivation out from the transient field: they're
not the same weight. The transient field still carries the derived projection
across phases (just via Catalyst's copy machinery instead of a tag). Lazy
derivation carries *nothing* new -- no tag, no field, no
`PlanAdaptiveSubqueries` hand-off -- and does strictly less optimizer work,
since `find` currently runs eagerly for every candidate DPP even though
projected reuse only matters after direct reuse fails.
On the transient point itself: I don't think it really applies on the
logical side. `sourcePlan` is a subtree of `buildQuery`, and `buildQuery` is a
plain, non-transient field on the same `DynamicPruningSubquery` -- it can be
non-transient because the node is `Unevaluable` and is rewritten into
`InSubqueryExec` before the plan is ever serialized for execution. So a
serialized plan would already retain `sourcePlan` through `buildQuery`; the
`@transient` on the projection doesn't add protection there. (It does genuinely
matter on `SubqueryAdaptiveBroadcastExec.buildPlan`, since that's a physical
node.)
All that said, transient second-parameter-list field is a perfectly good
outcome and clearly unblocks merge -- it keeps `productArity` at 7 and lets the
normal copy machinery carry the metadata, which removes the silent-drop risk. I
just wanted to make the lazy option explicit, since it's the lighter of the
two. Happy either way -- whatever you and @cloud-fan prefer.
--
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]
