HyukjinKwon commented on code in PR #59402:
URL: https://github.com/apache/spark/pull/59402#discussion_r4281845132
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala:
##########
@@ -290,9 +316,39 @@ object RewriteDistinctAggregates extends Rule[LogicalPlan]
{
case e => nullify(e)
}
+ // Union-of-filters guard for the shared distinct key. Without the
guard, the
+ // Expand projects the raw child for every row, so every distinct
child value
+ // becomes a key of the first aggregate, no matter how few rows the
filters
+ // select. Guarding the key with the union of the filters restores
the
+ // NULL-collapse: rows matching no predicate project NULL as the key
and merge
+ // into one key, keeping the first aggregate's output bounded by the
+ // pre-rewrite plan:
+ // distinct(child | Or(filters)) + 1 <= sum_i distinct(child |
filter_i) + n
+ // The guard must not apply when the group contains an unfiltered
aggregate
+ // (it needs the raw child over all rows), and the guard
re-evaluates the
+ // filters, so they must be deterministic. It also evaluates the
child only
+ // when the union is true, so every child must pass
+ // canEvaluateUnconditionally. A child that fails, such as a
division,
+ // stays raw and keeps today's evaluation count and error behavior.
It
+ // shares the rewriteCountDistinctConditional kill switch so
conf=false
+ // restores the pre-feature Expand key (raw child) for user-written
+ // FILTER distincts too.
+ val canGuard = SQLConf.get.rewriteCountDistinctConditionalEnabled &&
+ expressions.forall(_.filter.isDefined) &&
+ filters.forall(_.deterministic) &&
Review Comment:
**Non-blocking (P2):** `canGuard` treats any deterministic filter as safe to
evaluate a second time. But an `IN`, `NOT IN`, or `EXISTS` predicate subquery
is deterministic, and it has not been rewritten yet when this rule runs.
`RewritePredicateSubquery` runs later, in the `RewriteSubquery` batch, and
builds a separate `ExistenceJoin` for each occurrence in the Expand projection,
with no deduplication (`handleUnaryNode` -> `rewriteExistentialExprWithAttrs`).
With the guard on, each such filter appears twice in the projection: once as
its flag column and once inside `If(Or(filters), child, NULL)`. Every guarded
group with a subquery filter therefore pays one extra existence join over the
full input. For `NOT IN`, that is the null-aware join with an `OR ISNULL`
condition, which usually plans as a nested-loop join, so an O(N*M) join runs
twice.
With the default config this also hits existing user-written queries, not
just the new IF/CASE shapes. For example, `exists-outside-filter.sql` TC.01.06
(`sum(DISTINCT salary), count(DISTINCT hiredate) FILTER (WHERE EXISTS (SELECT 1
FROM dept WHERE emp.dept_id = dept.dept_id))`) joins `emp` with `dept` once at
the merge base and twice with this PR. Results don't change, so the result-only
goldens can't catch it. Two `COUNT(DISTINCT IF(b IN (SELECT ...), c, NULL))` on
one base reach the same shape through normalization.
Could `canGuard` also require every filter to be subquery-free
(`!_.containsPattern(PLAN_EXPRESSION)`, the exclusion
`canEvaluateUnconditionally` already applies to children)? `extractCondAndBase`
would also need to leave an IF/CASE whose condition contains a subquery
un-normalized. Otherwise those counts would collapse into a group that cannot
be guarded, which brings back the raw-key cardinality problem the guard exists
to prevent. With both exclusions, these shapes keep the merge-base plan.
**Verification:**
- **Regression:** A distinct FILTER with an EXISTS (and an IN) subquery
predicate yields exactly one existence join per subquery predicate in the
optimized plan with the conf on, the same count as with the conf off.
- **Regression:** Conditional distinct counts whose IF condition contains a
predicate subquery stay un-normalized, with no user-condition FILTER and the
same number of existence joins as the conf-off plan.
- **Behavior:** Results for the subquery-bearing queries match between conf
on and conf off.
- **Compatibility:** Existing union-of-filters guard shape, strict-reduction
gate, and raw-key fallback coverage remain green for subquery-free filters.
--
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]
