This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 0bf9defb1d fix: Fix three bugs in query decorrelation (#21208)
0bf9defb1d is described below
commit 0bf9defb1d9587ab2810e6e8adff512f4db58037
Author: Neil Conway <[email protected]>
AuthorDate: Tue Mar 31 09:15:15 2026 -0400
fix: Fix three bugs in query decorrelation (#21208)
## Which issue does this PR close?
- Closes #21205.
- Closes #21206.
- Closes #20315.
## Rationale for this change
This PR fixes three separate bugs in query decorrelation:
1. When removing duplicate filters as part of pulling up `IN`
subqueries, an operator precedence error meant that we would consider
two filters to be duplicates even if they involved different operators
(e.g., `=` and `>`).
2. When generating the `CASE` used to implement "count bug" handling, we
referenced the subquery output column without qualifying it by the
subquery alias. This could result in name-collisions with unrelated
identifiers in the parent query.
3. After generating the `CASE` used for "count bug" handling, we rewrote
the parent query to replace references to the subquery output column
with the generated `CASE` expression. This rewrite only matched on
unqualified column name only, which meant that unrelated parent query
identifiers that happened to share the same column name as subquery
aggregate aliases could been rewritten by mistake.
The first and third issues could result in incorrect query results; the
second would only cause spurious errors, as far as I can see.
## What changes are included in this PR?
* Fix all three bugs
* Add SLT tests
* Add a `debug_assert!` to document/check that
`remove_duplicated_filter` is called with a commutative operator
## Are these changes tested?
Yes.
## Are there any user-facing changes?
No, except in the sense that they fix user-visible bugs.
---
datafusion/optimizer/src/decorrelate.rs | 34 ++++++++---
.../optimizer/src/scalar_subquery_to_join.rs | 31 +++++-----
datafusion/sqllogictest/test_files/subquery.slt | 68 ++++++++++++++++++++++
3 files changed, 108 insertions(+), 25 deletions(-)
diff --git a/datafusion/optimizer/src/decorrelate.rs
b/datafusion/optimizer/src/decorrelate.rs
index e7bc62e8da..08839b49ef 100644
--- a/datafusion/optimizer/src/decorrelate.rs
+++ b/datafusion/optimizer/src/decorrelate.rs
@@ -26,7 +26,9 @@ use crate::simplify_expressions::ExprSimplifier;
use datafusion_common::tree_node::{
Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
TreeNodeRewriter,
};
-use datafusion_common::{Column, DFSchemaRef, HashMap, Result, ScalarValue,
plan_err};
+use datafusion_common::{
+ Column, DFSchemaRef, HashMap, Result, ScalarValue, assert_or_internal_err,
plan_err,
+};
use datafusion_expr::expr::Alias;
use datafusion_expr::simplify::SimplifyContext;
use datafusion_expr::utils::{
@@ -179,7 +181,7 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
find_join_exprs(subquery_filter_exprs)?;
if let Some(in_predicate) = &self.in_predicate_opt {
// in_predicate may be already included in the join
filters, remove it from the join filters first.
- join_filters = remove_duplicated_filter(join_filters,
in_predicate);
+ join_filters = remove_duplicated_filter(join_filters,
in_predicate)?;
}
let correlated_subquery_cols =
collect_subquery_cols(&join_filters, subquery_schema)?;
@@ -460,25 +462,39 @@ fn collect_local_correlated_cols(
}
}
-fn remove_duplicated_filter(filters: Vec<Expr>, in_predicate: &Expr) ->
Vec<Expr> {
- filters
+fn remove_duplicated_filter(
+ filters: Vec<Expr>,
+ in_predicate: &Expr,
+) -> Result<Vec<Expr>> {
+ // We assume below that swapping the order of operands to an operator does
+ // not change behavior, which is only true if the operator is commutative.
+ assert_or_internal_err!(
+ match in_predicate {
+ Expr::BinaryExpr(b) => b.op.swap() == Some(b.op),
+ _ => true,
+ },
+ "remove_duplicated_filter: in_predicate must use a commutative
operator"
+ );
+
+ Ok(filters
.into_iter()
.filter(|filter| {
if filter == in_predicate {
return false;
}
- // ignore the binary order
+ // Treat swapped operand order to a binary operator as equivalent
!match (filter, in_predicate) {
(Expr::BinaryExpr(a_expr), Expr::BinaryExpr(b_expr)) => {
- (a_expr.op == b_expr.op)
- && (a_expr.left == b_expr.left && a_expr.right ==
b_expr.right)
- || (a_expr.left == b_expr.right && a_expr.right ==
b_expr.left)
+ a_expr.op == b_expr.op
+ && ((a_expr.left == b_expr.left && a_expr.right ==
b_expr.right)
+ || (a_expr.left == b_expr.right
+ && a_expr.right == b_expr.left))
}
_ => false,
}
})
- .collect::<Vec<_>>()
+ .collect::<Vec<_>>())
}
fn agg_exprs_evaluation_result_on_empty_batch(
diff --git a/datafusion/optimizer/src/scalar_subquery_to_join.rs
b/datafusion/optimizer/src/scalar_subquery_to_join.rs
index 975c234b38..590b00098b 100644
--- a/datafusion/optimizer/src/scalar_subquery_to_join.rs
+++ b/datafusion/optimizer/src/scalar_subquery_to_join.rs
@@ -111,7 +111,7 @@ impl OptimizerRule for ScalarSubqueryToJoin {
// replace column references with entry in
map, if it exists
if let Some(map_expr) = expr
.try_as_col()
- .and_then(|col|
expr_check_map.get(&col.name))
+ .and_then(|col|
expr_check_map.get(col))
{
Ok(Transformed::yes(map_expr.clone()))
} else {
@@ -176,7 +176,7 @@ impl OptimizerRule for ScalarSubqueryToJoin {
// replace column references with entry in
map, if it exists
if let Some(map_expr) = expr
.try_as_col()
- .and_then(|col|
expr_check_map.get(&col.name))
+ .and_then(|col|
expr_check_map.get(col))
{
Ok(Transformed::yes(map_expr.clone()))
} else {
@@ -301,7 +301,7 @@ fn build_join(
subquery: &Subquery,
filter_input: &LogicalPlan,
subquery_alias: &str,
-) -> Result<Option<(LogicalPlan, HashMap<String, Expr>)>> {
+) -> Result<Option<(LogicalPlan, HashMap<Column, Expr>)>> {
let subquery_plan = subquery.subquery.as_ref();
let mut pull_up =
PullUpCorrelatedExpr::new().with_need_handle_count_bug(true);
let new_plan = subquery_plan.clone().rewrite(&mut pull_up).data()?;
@@ -358,14 +358,19 @@ fn build_join(
// If expr always returns null when column is null, skip
processing
continue;
}
+
+ let indicator_col =
+ Column::new(Some(subquery_alias), UN_MATCHED_ROW_INDICATOR);
+ // Qualify with the subquery alias to avoid ambiguity when the
+ // outer table has a column with the same name as the aggregate.
+ let value_col = Column::new(Some(subquery_alias), name.clone());
+
let computer_expr = if let Some(filter) =
&pull_up.pull_up_having_expr {
Expr::Case(expr::Case {
expr: None,
when_then_expr: vec![
(
- Box::new(Expr::IsNull(Box::new(Expr::Column(
-
Column::new_unqualified(UN_MATCHED_ROW_INDICATOR),
- )))),
+
Box::new(Expr::IsNull(Box::new(Expr::Column(indicator_col)))),
Box::new(result),
),
(
@@ -373,29 +378,23 @@ fn build_join(
Box::new(Expr::Literal(ScalarValue::Null, None)),
),
],
- else_expr:
Some(Box::new(Expr::Column(Column::new_unqualified(
- name.clone(),
- )))),
+ else_expr: Some(Box::new(Expr::Column(value_col.clone()))),
})
} else {
Expr::Case(expr::Case {
expr: None,
when_then_expr: vec![(
- Box::new(Expr::IsNull(Box::new(Expr::Column(
- Column::new_unqualified(UN_MATCHED_ROW_INDICATOR),
- )))),
+
Box::new(Expr::IsNull(Box::new(Expr::Column(indicator_col)))),
Box::new(result),
)],
- else_expr:
Some(Box::new(Expr::Column(Column::new_unqualified(
- name.clone(),
- )))),
+ else_expr: Some(Box::new(Expr::Column(value_col.clone()))),
})
};
let mut expr_rewrite = TypeCoercionRewriter {
schema: new_plan.schema(),
};
computation_project_expr
- .insert(name, computer_expr.rewrite(&mut
expr_rewrite).data()?);
+ .insert(value_col, computer_expr.rewrite(&mut
expr_rewrite).data()?);
}
}
diff --git a/datafusion/sqllogictest/test_files/subquery.slt
b/datafusion/sqllogictest/test_files/subquery.slt
index e5ca9d674e..7f88199b3c 100644
--- a/datafusion/sqllogictest/test_files/subquery.slt
+++ b/datafusion/sqllogictest/test_files/subquery.slt
@@ -1671,3 +1671,71 @@ drop table employees;
statement count 0
drop table project_assignments;
+
+# https://github.com/apache/datafusion/issues/21205
+statement ok
+CREATE TABLE dup_filter_t1(id INTEGER) AS VALUES (1), (2), (3);
+
+statement ok
+CREATE TABLE dup_filter_t2(id INTEGER) AS VALUES (1), (2), (3);
+
+query I
+SELECT * FROM dup_filter_t1 WHERE dup_filter_t1.id IN (
+ SELECT dup_filter_t2.id FROM dup_filter_t2 WHERE dup_filter_t2.id >
dup_filter_t1.id
+);
+----
+
+statement ok
+DROP TABLE dup_filter_t1;
+
+statement ok
+DROP TABLE dup_filter_t2;
+
+# https://github.com/apache/datafusion/issues/21206
+statement ok
+CREATE TABLE sq_name_t1(id INTEGER) AS VALUES (1), (2), (3);
+
+statement ok
+CREATE TABLE sq_name_t2(id INTEGER, outer_id INTEGER) AS VALUES (10, 1), (20,
1), (30, 2);
+
+query II
+SELECT sq_name_t1.id,
+ (SELECT count(*) AS id FROM sq_name_t2 WHERE sq_name_t2.outer_id =
sq_name_t1.id) AS cnt
+FROM sq_name_t1
+ORDER BY sq_name_t1.id;
+----
+1 2
+2 1
+3 0
+
+query I
+SELECT sq_name_t1.id
+FROM sq_name_t1
+WHERE sq_name_t1.id > (
+ SELECT count(*) AS id
+ FROM sq_name_t2
+ WHERE sq_name_t2.outer_id = sq_name_t1.id
+)
+ORDER BY sq_name_t1.id;
+----
+2
+3
+
+query I
+SELECT sq_name_t1.id * 10 + (
+ SELECT count(*) AS id
+ FROM sq_name_t2
+ WHERE sq_name_t2.outer_id = sq_name_t1.id
+) AS total
+FROM sq_name_t1
+ORDER BY sq_name_t1.id;
+----
+12
+21
+30
+
+statement ok
+DROP TABLE sq_name_t1;
+
+statement ok
+DROP TABLE sq_name_t2;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
