You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/06 06:10:10 UTC
(datafusion) 01/01: fix: keep filters with correlated subqueries above extension nodes (#25294)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25294-c0e872fb4d02542cfa86b7bc40535ee5023a325a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 8f4c51fdf2246e2ff802d50d5f2bc290c25e4f28
Author: Rusty Raven <ka...@gmail.com>
AuthorDate: Tue Oct 6 06:09:42 2026 +0000
fix: keep filters with correlated subqueries above extension nodes (#25294)
## Which issue does this PR close?
- Closes #15046.
## Rationale for this change
An extension node can name columns whose predicates must not be pushed
past it, via
`UserDefinedLogicalNode::prevent_predicate_push_down_columns`. A
predicate whose only dependency on those columns runs through a
correlated subquery is pushed down anyway, and the resulting plan does
not just perform worse — it fails to execute.
Using the `NoopPlan` node already present in the optimizer's own tests,
which refuses push-down of column `c`:
```
Filter: EXISTS (SELECT sq.a FROM sq WHERE outer_ref(test.c) = sq.a)
NoopPlan
TableScan: test
```
optimizing that plan today gives
```
Optimizer rule 'push_down_filter' failed
Invalid (non-executable) plan after Optimizer rule: push_down_filter
In/Exist/SetComparison subquery can only be used in Projection, Filter, TableScan,
Window functions, Aggregate and Join plan nodes, but was used in [NoopPlan]
```
The predicate is moved below the node that asked to keep it, lands
somewhere subqueries are not permitted, and the invariant check turns
that into a query failure.
The cause is the one #15046 describes. The check is
```rust
!expr.column_refs().iter().any(|c| prevent_cols.contains(&c.name))
```
and `column_refs` collects only `Expr::Column`. A subquery records the
outer columns it correlates on in `Subquery::outer_ref_columns`, which
`Expr`'s traversal does not descend into, so `EXISTS (... WHERE outer.c
= ...)` reports no column references at all and looks unconditionally
safe to push.
## What changes are included in this PR?
A `references_any_column` helper in `push_down_filter.rs` that asks
`column_refs` first and then, for `Expr::Exists`, `Expr::InSubquery` and
`Expr::ScalarSubquery`, also looks at the `outer_ref_columns` those
carry. The extension-node branch uses it instead of `column_refs` alone.
Nothing else changes.
**I deliberately did not change `column_refs` itself**, which is what
the issue title asks for. It has many callers whose behaviour would
shift — `optimize_projections`, CSE, window and join handling all use
it, and several of them want "columns produced by this plan's own
schema", not outer references belonging to a parent. Widening it looks
like a separate change with its own risk, so this PR fixes the consumer
that has a demonstrable failure. Happy to look at the broader change
instead if you would rather that were the fix.
## What is the testing strategy for this PR?
New unit test
`push_down_filter::tests::user_defined_plan_outer_referenced_column`,
next to the existing `user_defined_plan` test and reusing its
`NoopPlan`. It asserts the `EXISTS` predicate stays above the node.
The test discriminates: with the outer-reference collection
short-circuited out, it fails with the invalid-plan error quoted above;
with the fix it passes. `datafusion-optimizer` (808) and
`datafusion-expr` (262) pass, as does `sqllogictest`.
## Are there any user-facing changes?
No public API changes. Plans change only where an extension node
declares `prevent_predicate_push_down_columns` and a predicate
correlates a subquery on one of those columns — previously an invalid
plan, now a correct one. No in-tree node declares those columns outside
tests, so no bundled plan snapshots move.
Two adjacent things found while working on this, neither touched here:
- `find_out_reference_exprs` does not see these outer references either,
for the same reason, so it cannot be used as the collection mechanism.
That is the same traversal gap as #16147, which @AdamGS has taken.
- The equivalent guard for correlated subqueries in other push-down
branches was not audited; this PR only covers the extension-node branch
that #15046 reports.
---
datafusion/optimizer/src/push_down_filter.rs | 144 ++++++++++++++++++++++++---
1 file changed, 132 insertions(+), 12 deletions(-)
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
index b358c8abd2..ab7a2d26b5 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -26,7 +26,9 @@ use itertools::Itertools;
use log::{Level, debug, log_enabled};
use datafusion_common::instant::Instant;
-use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::tree_node::{
+ Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
+};
use datafusion_common::{
Column, DFSchema, Result, assert_eq_or_internal_err, internal_err, plan_err,
qualified_name,
@@ -783,6 +785,45 @@ fn infer_join_predicates_impl<
Ok(())
}
+/// Whether `expr` depends on any of the columns named in `names`.
+///
+/// This is the columns `Expr::column_refs` would collect plus the outer columns
+/// that any subquery inside `expr` correlates on. A subquery records those in
+/// `Subquery::outer_ref_columns` rather than as an `Expr::Column` in the
+/// predicate, and `Expr`'s own traversal does not descend into that field, so
+/// looking only at `column_refs` would report such a predicate as depending on
+/// nothing and let it be pushed past a node that asked to keep those columns.
+fn references_any_column(expr: &Expr, names: &HashSet<String>) -> bool {
+ let mut found = false;
+ expr.apply(|e| {
+ let outer_refs = match e {
+ Expr::Column(col) => {
+ if names.contains(&col.name) {
+ found = true;
+ return Ok(TreeNodeRecursion::Stop);
+ }
+ return Ok(TreeNodeRecursion::Continue);
+ }
+ Expr::Exists(exists) => &exists.subquery.outer_ref_columns,
+ Expr::InSubquery(in_subquery) => &in_subquery.subquery.outer_ref_columns,
+ Expr::ScalarSubquery(subquery) => &subquery.outer_ref_columns,
+ Expr::SetComparison(set_comparison) => {
+ &set_comparison.subquery.outer_ref_columns
+ }
+ _ => return Ok(TreeNodeRecursion::Continue),
+ };
+ if outer_refs.iter().any(|outer_ref| {
+ matches!(outer_ref, Expr::OuterReferenceColumn(_, c) if names.contains(&c.name))
+ }) {
+ found = true;
+ return Ok(TreeNodeRecursion::Stop);
+ }
+ Ok(TreeNodeRecursion::Continue)
+ })
+ .expect("traversal is infallible");
+ found
+}
+
impl OptimizerRule for PushDownFilter {
fn name(&self) -> &str {
"push_down_filter"
@@ -1301,12 +1342,7 @@ impl OptimizerRule for PushDownFilter {
let predicate_push_or_keep: Vec<bool> =
split_conjunction(&filter.predicate)
.iter()
- .map(|expr| {
- !expr
- .column_refs()
- .iter()
- .any(|c| prevent_cols.contains(&c.name))
- })
+ .map(|expr| !references_any_column(expr, &prevent_cols))
.collect();
// all predicates are kept, no changes needed
@@ -1523,14 +1559,14 @@ mod tests {
use arrow::datatypes::{Field, Metadata, Schema, SchemaRef};
use async_trait::async_trait;
- use datafusion_common::{DFSchemaRef, DataFusionError, ScalarValue};
- use datafusion_expr::expr::ScalarFunction;
+ use datafusion_common::{DFSchemaRef, DataFusionError, ScalarValue, Spans};
+ use datafusion_expr::expr::{ScalarFunction, SetComparison, SetQuantifier};
use datafusion_expr::logical_plan::table_scan;
use datafusion_expr::{
ColumnarValue, ExprFunctionExt, Extension, LogicalPlanBuilder,
- ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, TableScan, TableSource,
- TableType, UserDefinedLogicalNodeCore, Volatility, WindowFunctionDefinition, col,
- in_list, in_subquery, lit,
+ ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Subquery, TableScan,
+ TableSource, TableType, UserDefinedLogicalNodeCore, Volatility,
+ WindowFunctionDefinition, col, exists, in_list, in_subquery, lit, out_ref_col,
};
use crate::OptimizerContext;
@@ -2185,6 +2221,90 @@ mod tests {
}
}
+ #[test]
+ fn user_defined_plan_outer_referenced_column() -> Result<()> {
+ let table_scan = test_table_scan()?;
+
+ // A subquery correlated on `test.c` — the column `NoopPlan` refuses to
+ // have predicates pushed past. The correlation is carried by the
+ // subquery's `outer_ref_columns`, not by an `Expr::Column` in the
+ // predicate itself.
+ let subquery = LogicalPlanBuilder::from(test_table_scan_with_name("sq")?)
+ .filter(out_ref_col(DataType::UInt32, "test.c").eq(col("sq.a")))?
+ .project(vec![col("sq.a")])?
+ .build()?;
+
+ let custom_plan = LogicalPlan::Extension(Extension {
+ node: Arc::new(NoopPlan {
+ input: vec![table_scan.clone()],
+ schema: Arc::clone(table_scan.schema()),
+ }),
+ });
+ let plan = LogicalPlanBuilder::from(custom_plan)
+ .filter(exists(Arc::new(subquery)))?
+ .build()?;
+
+ // The predicate depends on `test.c`, so it must stay above NoopPlan.
+ assert_optimized_plan_equal!(
+ plan,
+ @r"
+ Filter: EXISTS (<subquery>)
+ Subquery:
+ Projection: sq.a
+ TableScan: sq, full_filters=[outer_ref(test.c) = sq.a]
+ NoopPlan
+ TableScan: test
+ "
+ )
+ }
+
+ #[test]
+ fn user_defined_plan_outer_referenced_column_set_comparison() -> Result<()> {
+ let table_scan = test_table_scan()?;
+
+ // `test.a > ANY (SELECT sq.a FROM sq WHERE test.c = sq.a)`: the
+ // comparison expression names only `test.a`, so the dependency on
+ // `test.c` exists solely in the subquery's `outer_ref_columns`.
+ let subquery = LogicalPlanBuilder::from(test_table_scan_with_name("sq")?)
+ .filter(out_ref_col(DataType::UInt32, "test.c").eq(col("sq.a")))?
+ .project(vec![col("sq.a")])?
+ .build()?;
+ let outer_ref_columns = subquery.all_out_ref_exprs();
+ let set_comparison = Expr::SetComparison(SetComparison::new(
+ Box::new(col("test.a")),
+ Subquery {
+ subquery: Arc::new(subquery),
+ outer_ref_columns,
+ spans: Spans::new(),
+ },
+ Operator::Gt,
+ SetQuantifier::Any,
+ ));
+
+ let custom_plan = LogicalPlan::Extension(Extension {
+ node: Arc::new(NoopPlan {
+ input: vec![table_scan.clone()],
+ schema: Arc::clone(table_scan.schema()),
+ }),
+ });
+ let plan = LogicalPlanBuilder::from(custom_plan)
+ .filter(set_comparison)?
+ .build()?;
+
+ // The predicate depends on `test.c`, so it must stay above NoopPlan.
+ assert_optimized_plan_equal!(
+ plan,
+ @r"
+ Filter: test.a > ANY (<subquery>)
+ Subquery:
+ Projection: sq.a
+ TableScan: sq, full_filters=[outer_ref(test.c) = sq.a]
+ NoopPlan
+ TableScan: test
+ "
+ )
+ }
+
#[test]
fn user_defined_plan() -> Result<()> {
let table_scan = test_table_scan()?;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org