starocean999 opened a new pull request, #67940:
URL: https://github.com/apache/doris/pull/67940
Problem Summary:
A subquery which aggregates may filter its own aggregation with a HAVING
clause, and the rows which that clause removes have to be handled by the plan
which executes the subquery. Three groups of wrong results around that pattern
are fixed here (all of them silent wrong results, without any error):
1. A correlated `EXISTS`/`NOT EXISTS` subquery which aggregates
(`UnCorrelatedApplyAggregateFilter`). Unnesting it moves the correlated
predicate into the join and appends the inner side of the predicate to the
group by keys of the aggregate, which is only equivalent to the subquery when
the correlated predicate is an equality between the outer side and the inner
side (then the inner rows of one outer row are exactly one group) and when the
aggregate does not return a row which has to survive an empty domain. Three
shapes returned a wrong result:
* a non equality correlated predicate with a group by and a HAVING
clause. With `o = (3)` and `i = (1, 10), (2, 10)`, `select o.k from o where
exists (select count(*) from i where i.k < o.k group by i.g having count(*) =
2)` returned an empty result instead of `3`, and the same query with `not
exists` returned `3` instead of an empty result: `having count(*) = 2` was
evaluated on the groups `(i.k, i.g)` instead of on the domain `i.k < o.k`
grouped by `i.g`.
* a global aggregate (no group by) whose HAVING clause holds for an empty
input. A global aggregate returns exactly one row for every outer row,
including the outer rows without any matching inner row; as soon as the inner
side of the correlated predicate becomes the group by key, that row does not
exist any more. With `o = (1), (2)` and `i = (2)`, `select o.k from o where
exists (select count(*) from i where i.k = o.k having count(*) = 0)` returned
an empty result instead of `1`, and the same query with `not exists` returned
`1, 2` instead of `2`.
* a HAVING clause which references the outer query.
`UnCorrelatedApplyFilter` pulls the correlated HAVING predicates into the
apply, and the analyzer can leave a passthrough projection of the select list
above the aggregate (`Project(count(*)) over Aggregate(...)`), so the apply is
`Apply(Project(Aggregate(Filter)))`, a shape the rule did not match: the
correlation was dropped and the query degraded to a global aggregate joined on
the HAVING predicate. With `o = (1), (2), (7), (NULL), (-1)` and `i = (1, 10),
(2, 10), (5, 20), (NULL, 30)`, `select o.k from o where exists (select count(*)
from i where i.k = o.k having count(*) <= o.k - 7)` returned an empty result
instead of `7`.
2. `x [not] in (aggregating subquery)` used as a value (in the select list,
in a comparison, ...) when the subquery has a HAVING clause which references
the outer row. `InApplyToJoin` builds the plan of such a subquery as a mark
join whose equality with the subquery output is the only hash condition and
whose other join conjuncts are the predicates of the HAVING clause, and
`PhysicalPlanTranslator` reuses that equality as the hash condition of a null
aware semi/anti join when the hash join conjuncts are empty. With `o = (3),
(5), (8), (NULL)` and `i = (1, 10), (2, 10), (5, 20), (NULL, 30)` (the
aggregation `count(*)` of `i` is the folded constant 4),
select o.k, o.k in (select count(*) from i having o.k <= 8) as m from
o order by o.k
returned `m = NULL` for `k = 3` (the HAVING clause holds there, and `3 IN
(4)` is `FALSE`) instead of `FALSE`, and the same query with `not in` returned
`NULL` instead of `TRUE`; the plan was
3:VHASH JOIN
| join op: NULL AWARE LEFT SEMI JOIN(BROADCAST)
| other join predicates: (k <= 8)
| mark join predicates: (expr_cast(k as BIGINT) = count(*))
| final projections: k, $c$1
3. An uncorrelated scalar subquery which aggregates and whose HAVING clause
removes the row of the aggregation. With `sq_o = (7, 2)` and `sq_i = (1)`,
SELECT o.id, (SELECT COUNT(*) FROM sq_i WHERE k > 0 HAVING
COUNT(*)>1) AS s
FROM sq_o o ORDER BY o.id
returned an empty result instead of `7 | NULL`: a scalar subquery is
planned by `ScalarApplyToJoin` as a `CROSS JOIN` with a
`LogicalAssertNumRows(EQ 1)` which both checks that the subquery returns at
most one row and builds the single null row of an empty subquery (the backend
`AssertNumRowsOperatorX` inserts that row when its input is empty), and that
assertion was eliminated.
Fix:
1. When the original rewrite of `EXISTS`/`NOT EXISTS` is not equivalent, the
subquery is rewritten by
`UnCorrelatedApplyAggregateFilter#pullUpCorrelatedPredicateByAggregatingOuter`,
which computes the aggregation together with the outer side: the distinct
correlation keys of a deep copy of the outer plan (so that repeated outer
values do not multiply the aggregates) are joined with the inner side on the
predicates of the WHERE clause and of the pulled domain predicates, with the
outer side replaced by the correlation key (a left outer join when the
aggregate has no group by, so that an empty correlated domain keeps one row, an
inner join otherwise); the aggregate is computed for `(correlation key, group
by key)` with a projected marker column which tells whether the inner side
matched the correlated predicate, so that `count(*)` becomes `count(marker)`
and `count(argument)` becomes `count(if(marker, argument, null))`; the
predicates which were pulled out of the HAVING clause are c
lassified by their provenance (a predicate which references the output of the
aggregate is evaluated above the new aggregate on the group of the outer row
with its outer slots replaced by the correlation keys, a predicate of the WHERE
clause stays a join condition); and the outer rows are filtered by a `LEFT
SEMI/ANTI JOIN` whose condition is the null safe equality of the outer
correlation slots and the aggregated keys. The rule also matches the shapes
where the aggregation of the subquery is below the projection of its select
list and where predicates of the HAVING clause which do not reference the outer
query stay below that projection. The original rewrite is kept whenever it is
equivalent and for the shapes the new path cannot handle safely.
2. `be/src/exec/common/hash_table/join_hash_table.h`: for a null aware
semi/anti join with other join conjuncts the probe operator keeps the null
flags of the search (`_null_flags`) and
`ProcessHashTableProbe::do_mark_join_conjuncts` copies them as a whole into the
null map of the mark column, so every row of the batch must have its flag
written by the search. The placeholder row which
`_find_null_aware_with_other_conjuncts_impl` emits for a probe row without any
match (`build_idxs[matched_cnt] = 0`) did not write it, and `_null_flags` is
reused by the following batches without being cleared, so that row inherited
the flag of a previous batch: a stale true flag (written when a probe row with
a null key matched build rows) was read as "matched a null" and turned the mark
of the row into null. The placeholder row now writes the flag as false (like
the placeholder row of `_process_probe_null_key` does); the null case is still
represented by the rows which matched null keys of the bui
ld side, and the plan does not change.
3. `EliminateAssertNumRows`: for `Assertion.EQ` with `desiredNumOfRows == 1`
it returned true as soon as the plan below the skipped nodes is an aggregation
without a group by, because such an aggregation always returns exactly one row.
The nodes it skips include `LogicalFilter` and the preserved side of a
semi/anti join, which only keep the rows which satisfy them: a HAVING clause
above the aggregation can remove its single row, and then the assertion is the
operator which turns the empty subquery into its null value. The rule now
remembers whether the nodes it skips over may reduce the number of rows of the
checked plan (`mayReduceRowCount`: a `LogicalFilter` or a left/right semi/anti
join) and the `EQ 1` shortcut is only taken when the row count is preserved
(only projections and sorts were skipped). The upper bound based elimination of
the other assertions is unchanged, because the skipped nodes can only make the
bound more conservative.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should
merge into -->
--
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]
