Improve error for unsupported correlated predicate

In the error message report correlated part besides the full predicate.

datafusion/optimizer/src/analyzer/subquery.rs

@@ -21,7 +21,7 @@ use crate::utils::collect_subquery_cols;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
 use datafusion_common::{plan_err, Result};
 use datafusion_expr::expr_rewriter::strip_outer_reference;
-use datafusion_expr::utils::split_conjunction;
+use datafusion_expr::utils::{conjunction, split_conjunction};
 use datafusion_expr::{Aggregate, Expr, Filter, Join, JoinType, LogicalPlan, Window};
 
 /// Do necessary check on subquery expressions and fail the invalid plan
@@ -139,8 +139,48 @@ fn check_inner_plan(inner_plan: &LogicalPlan, can_contain_outer_ref: bool) -> Re
             })?;
             Ok(())
         }
+<<<<<<< HEAD
         LogicalPlan::Filter(Filter { input, .. }) => {
             check_inner_plan(input, can_contain_outer_ref)
+||||||| parent of 44bfe0d0b (Improve error for unsupported correlated predicate)
+        LogicalPlan::Filter(Filter {
+            predicate, input, ..
+        }) => {
+            let (correlated, _): (Vec<_>, Vec<_>) = split_conjunction(predicate)
+                .into_iter()
+                .partition(|e| e.contains_outer());
+            let maybe_unsupported = correlated
+                .into_iter()
+                .filter(|expr| !can_pullup_over_aggregation(expr))
+                .collect::<Vec<_>>();
+            if is_aggregate && is_scalar && !maybe_unsupported.is_empty() {
+                return plan_err!(
+                    "Correlated column is not allowed in predicate: {predicate}"
+                );
+            }
+            check_inner_plan(input, is_scalar, is_aggregate, can_contain_outer_ref)
+=======
+        LogicalPlan::Filter(Filter {
+            predicate, input, ..
+        }) => {
+            let (correlated, _): (Vec<_>, Vec<_>) = split_conjunction(predicate)
+                .into_iter()
+                .partition(|e| e.contains_outer());
+            let maybe_unsupported = correlated
+                .into_iter()
+                .filter(|expr| !can_pullup_over_aggregation(expr))
+                .collect::<Vec<_>>();
+            if is_aggregate && is_scalar && !maybe_unsupported.is_empty() {
+                let unsupported_for_display =
+                    conjunction(maybe_unsupported.iter().map(|expr| (*expr).to_owned()))
+                        // safe to unwrap because maybe_unsupported is not empty
+                        .unwrap();
+                return plan_err!(
+                    "Correlated expression is not allowed in filter: {unsupported_for_display} (full predicate: {predicate})"
+                );
+            }
+            check_inner_plan(input, is_scalar, is_aggregate, can_contain_outer_ref)
+>>>>>>> 44bfe0d0b (Improve error for unsupported correlated predicate)
         }
         LogicalPlan::Window(window) => {
             check_mixed_out_refer_in_window(window)?;