jayzhan211 commented on code in PR #25338:
URL: https://github.com/apache/datafusion/pull/25338#discussion_r4105298413
##########
datafusion/optimizer/src/decorrelate.rs:
##########
@@ -182,6 +195,11 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
.iter()
.filter(|e| e.contains_outer())
.all(|&e| can_pullup_over_aggregation(e));
+ for expr in &subquery_filter_exprs {
+ if expr.contains_outer() &&
!self.correlated_filters.contains(expr) {
+ self.correlated_filters.push((*expr).clone());
Review Comment:
`correlated_filters` records a conjunct regardless of what sits above the
`Filter`. A LEFT JOIN (filter on the nullable side) or a `ROLLUP` above it puts
NULLs back into the key, but `filter_rejects_null` still reads `u.y = o.x` as
"never NULL", so the `NOT IN` anti join loses `null_aware`. `main` returns the
correct result for both queries below:
```sql
CREATE TABLE o(x INT) AS VALUES (1),(3),(NULL);
CREATE TABLE t(k INT) AS VALUES (1),(2);
CREATE TABLE u(y INT, k INT) AS VALUES (1,1);
SELECT x FROM o WHERE x NOT IN (
SELECT u.y FROM t LEFT JOIN (SELECT * FROM u WHERE u.y = o.x) AS u ON t.k
= u.k);
-- expected (and main): no rows; this PR: 3, NULL
-- co/ci from subquery_projection.slt
SELECT co.id, co.k FROM co WHERE co.k NOT IN (
SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY ROLLUP(ci.k));
-- expected (and main): no rows; this PR: (NULL,3), (9,9), (5,NULL)
```
Fix: drop the recorded filters when the pull up passes such a node, and add
both queries to the slt:
```diff
fn f_up(&mut self, plan: LogicalPlan) ->
Result<Transformed<LogicalPlan>> {
+ // A node that null-extends or regroups rows can put a NULL back
into
+ // a column that a correlated filter below it rejected.
+ if may_reintroduce_nulls(&plan) {
+ self.correlated_filters.clear();
+ }
let subquery_schema = plan.schema();
```
```rs
fn may_reintroduce_nulls(plan: &LogicalPlan) -> bool {
match plan {
LogicalPlan::Join(join) => matches!(
join.join_type,
JoinType::Left | JoinType::Right | JoinType::Full
),
LogicalPlan::Union(_) => true,
LogicalPlan::Aggregate(aggregate) => aggregate
.group_expr
.iter()
.any(|e| matches!(e, Expr::GroupingSet(_))),
_ => false,
}
}
```
--
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]
