@@ -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()?;