pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4278473903
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -158,20 +159,23 @@ object RewritePredicateSubquery extends Rule[LogicalPlan]
with PredicateHelper {
// Filter the plan by applying left semi and left anti joins.
withSubquery.foldLeft(newFilter) {
case (p, Exists(sub, _, _, conditions, subHint)) =>
- val (joinCond, outerPlan) = rewriteExistentialExpr(conditions, p)
- val join = buildJoin(outerPlan,
rewriteDomainJoinsIfPresent(outerPlan, sub, joinCond),
+ val (joinCond, outerPlan, newSub) =
+ rewriteExistentialExprInJoinCondition(conditions, p, sub)
+ val join = buildJoin(outerPlan,
rewriteDomainJoinsIfPresent(outerPlan, newSub, joinCond),
LeftSemi, joinCond, subHint)
Project(p.output, join)
case (p, Not(Exists(sub, _, _, conditions, subHint))) =>
- val (joinCond, outerPlan) = rewriteExistentialExpr(conditions, p)
- val join = buildJoin(outerPlan,
rewriteDomainJoinsIfPresent(outerPlan, sub, joinCond),
+ val (joinCond, outerPlan, newSub) =
+ rewriteExistentialExprInJoinCondition(conditions, p, sub)
+ val join = buildJoin(outerPlan,
rewriteDomainJoinsIfPresent(outerPlan, newSub, joinCond),
LeftAnti, joinCond, subHint)
Project(p.output, join)
case (p, InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))) =>
// Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- val (joinCond, outerPlan) = rewriteExistentialExpr(inConditions ++
conditions, p)
+ val dedupSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))
+ val inConditions = values.zip(dedupSub.output).map(EqualTo.tupled)
+ val (joinCond, outerPlan, newSub) =
+ rewriteExistentialExprInJoinCondition(inConditions ++ conditions,
p, dedupSub)
Review Comment:
Done, in a separate commit on this branch so the routing fix and the
estimation change stay separately reviewable.
### Both statistics modes
`SizeInBytesOnlyStatsPlanVisitor.visitJoin` gains an `ExistenceJoin` arm
that rescales the left child's size the same way `visitUnaryNode` does, since
the row count is unchanged and only the appended Boolean widens each row:
```scala
case ExistenceJoin(_) =>
visitExistenceJoin(p)
```
`JoinEstimation` gains `estimateExistenceJoin()`, which takes the row count
from the left child and sizes the output from `join.output` — that is the left
output plus `exists`, so the appended width is accounted for, and the attribute
has no column statistics so `getSizePerRow` falls back to its type's default
size. Only the left child's attribute statistics are propagated; the right
child's describe attributes an existence join does not output.
It requires only `rowCountsExist(join.left)`, so a right child with no
statistics, or an empty one, is still estimated rather than falling back to the
product.
### Measured
On the shape from this thread, with `spark.sql.cbo.enabled` either way:
| | left | right | node estimate |
|---|---|---|---|
| before | 2,579 | 2,591 | **6,682,189** |
| after | 2,579 | 2,591 | **2,601** |
The old estimate is the product exactly — a 2,591x over-estimate, the right
child's size used as a multiplier — for an operator that emits precisely the
left child's 2,579 bytes plus one byte per row.
### A TPC-DS plan actually improves
`q10.sf100` under `TPCDSV1_4_PlanStabilityWithStatsSuite` moves, and it is
the regression you predicted, in an approved plan. Net operator change:
```
-1 SortMergeJoin +1 BroadcastHashJoin
-2 Exchange +1 BroadcastExchange
-2 Sort
-3 WholeStageCodegen
```
The subtree carrying `Filter [exists,exists]` is now small enough to
broadcast, so the shuffle and sort of the customer-side input disappear:
```
- SortMergeJoin [c_current_cdemo_sk,cd_demo_sk]
- ... Sort [c_current_cdemo_sk]
- ... Exchange [c_current_cdemo_sk] #2
+ BroadcastHashJoin [c_current_cdemo_sk,cd_demo_sk]
+ ... BroadcastExchange #2
```
Golden files regenerated with `SPARK_GENERATE_GOLDEN_FILES=1`, not edited.
This is the only one of the 8 approved plans containing an `ExistenceJoin` that
changes, and only in the CBO-enabled suite — which fits, since that is the mode
where `JoinEstimation` used to return `None`.
### Regression
The new plan-selection test in `SubquerySuite` is the one you asked for: a
nested inner-routed existence join whose left child is below the broadcast
threshold while the product would exceed it, asserting the enclosing equi semi
join keeps its broadcast. It is marked `DisableAdaptiveExecution`, since with
AQE on runtime statistics recover the plan and the estimate under test is not
the one the planner uses. It also notes why the hoisted predicate is `a > c1`
rather than the key equality: with a disjunction as the whole condition the
join is a broadcast nested loop join whatever the statistics say, so it could
not show the effect.
Checked that it discriminates — with the two new arms reverted it fails with
```
6682189 was not less than 5158 the existence join is estimated at 6682189
from a
left child of 2579 and a right child of 2591
```
Statistics regressions cover both modes: `existence join` and `existence
join with an empty or unestimated right child` in `JoinEstimationSuite` pin the
CBO row count, size and left-only attribute stats, including a right child with
a zero row count and one with no column statistics; `existence join is
estimated from its left child in both statistics modes` in
`BasicStatsEstimationSuite` pins the same invariant through `checkStats` with
CBO on and off, and asserts the estimate is not the product.
--
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]
