peter-toth commented on code in PR #57437:
URL: https://github.com/apache/spark/pull/57437#discussion_r3682057237
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -127,23 +127,36 @@ case class InSubqueryExec(
override def nullable: Boolean = child.nullable
override def toString: String = s"$child IN ${plan.name}"
- override def withNewPlan(plan: BaseSubqueryExec): InSubqueryExec = copy(plan
= plan)
+ override def withNewPlan(plan: BaseSubqueryExec): InSubqueryExec =
+ copy(plan = plan, resultBroadcast = null, result = null)
Review Comment:
**Finding 2.** Clearing `result` is what this PR needs; clearing
`resultBroadcast` reaches a path the PR isn't targeting.
`resultBroadcast` is only ever assigned for non-DPP subqueries —
`updateResult` guards it with `!isDynamicPruning && !isResultUnavailable` — so
the unavailable marker can only ever live in `result`, and for DPP
`resultBroadcast` is always null anyway. The one case where the extra reset
does something is an ordinary `IN` subquery: `ReuseAdaptiveSubquery` calls
`withNewPlan(ReusedSubqueryExec(...))`, and if that lands after
`updateResult()` has run, the new instance now drops a broadcast that base kept
and has to re-run `updateResult()` and re-broadcast. It recovers (a fresh
instance re-`prepare`s), so nothing breaks — but it's a silent behavior change
to a shared method, with no test and nothing in the description.
```suggestion
copy(plan = plan, result = null)
```
If clearing the broadcast is deliberate — say you want a plan swap to always
recompute — a one-line comment saying so would stop the next reader from
"simplifying" it back.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/DynamicPruning.scala:
##########
@@ -80,7 +86,20 @@ case class DynamicPruningSubquery(
// DynamicPruningSubquery should only have a single broadcasting key
since
// there are no usage for multiple broadcasting keys at the moment.
broadcastKeyIndices.size == 1 &&
- child.dataType == buildKeys(broadcastKeyIndices.head).dataType
+ child.dataType == buildKeys(broadcastKeyIndices.head).dataType &&
+ broadcastValueProjection.forall { projection =>
Review Comment:
**Finding 3.** This is the one place in the PR where a bad projection fails
the query rather than being discarded.
Every clause here is already guaranteed by
`ReusableBroadcastValueProjection.find`:
- `sourcePlan.deterministic`, `sourceHashKeys.nonEmpty`, and the hash-key
determinism come from the `candidate.filter` block plus `isSafeSourceHashKey`;
- `sourceHashKeys` / `valueExpression`
`references.subsetOf(sourcePlan.outputSet)` come from `ExtractEquiJoinKeys`'
`canEvaluate` and from `find`'s own `value.references.subsetOf(left.outputSet)`;
- `valueExpression.deterministic` comes from `isSafeValueExpression`;
- `valueExpression.dataType == pruningKey.dataType` follows because the
value starts as `filteringKeys.head`, `PartitionPruning` sets `indices =
Seq(joinKeys.indexOf(filteringKeys.head))`, and `replaceAlias` is
type-preserving — so it is the same expression the line above already
type-checks against `pruningKey`.
So these clauses can only fire for a projection that did not come from
`find`. When they do, the effect is `resolved == false` on a node inside an
already-optimized plan: the containing `Filter` reports unresolved,
`LogicalPlanIntegrity.validateOptimizedPlan` flags it under plan-change
validation, and outside validation an unresolved node just keeps flowing. That
inverts the contract the description states — "If Spark cannot verify the
broadcast, cannot safely evaluate the expression ... it discards the entire
projected domain and continues without this optimization." Your own test pins
the fail-closed shape:
```scala
assert(!pruning.copy()(Some(projection.copy(sourceHashKeys =
Seq(missing)))).resolved)
```
Keeping the checks but not the hard failure, either:
```scala
// drop rather than un-resolve; both planner rules read this instead of the
raw field
private[sql] lazy val usableBroadcastValueProjection:
Option[BroadcastValueProjection] =
broadcastValueProjection.filter(isSafeProjection)
```
or move them to a `require` in `BroadcastValueProjection`'s producer, so a
malformed projection is a bug at the producer instead of a plan-validation
failure later. Either way `DynamicPruningSubquery.resolved` stays about the DPP
itself, and an unusable projection behaves like every other failure in this
feature.
--
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]
