[WIP] refactor: framework for subquery decorrelation

datafusion/expr/src/expr.rs

@@ -1655,6 +1655,25 @@ impl Expr {
         using_columns
     }
 
+    pub fn outer_column_refs(&self) -> HashSet<&Column> {
+        let mut using_columns = HashSet::new();
+        self.add_outer_column_refs(&mut using_columns);
+        using_columns
+    }
+
+    /// Adds references to all outer columns in this expression to the set
+    ///
+    /// See [`Self::column_refs`] for details
+    pub fn add_outer_column_refs<'a>(&'a self, set: &mut HashSet<&'a Column>) {
+        self.apply(|expr| {
+            if let Expr::OuterReferenceColumn(_, col) = expr {
+                set.insert(col);
+            }
+            Ok(TreeNodeRecursion::Continue)
+        })
+        .expect("traversal is infallible");
+    }
+
     /// Adds references to all columns in this expression to the set
     ///
     /// See [`Self::column_refs`] for details
@@ -1715,6 +1734,19 @@ impl Expr {
             .expect("exists closure is infallible")
     }
 
+    /// Return true if the expression contains out reference(correlated) expressions.
+    pub fn contains_outer_from_relation(&self, outer_relation_name: &String) -> bool {
+        self.exists(|expr| {
+            if let Expr::OuterReferenceColumn(_, col) = expr {
+                if let Some(relation) = &col.relation {
+                    return Ok(relation.table() == outer_relation_name);
+                }
+            }
+            Ok(false)
+        })
+        .expect("exists closure is infallible")
+    }
+
     /// Returns true if the expression node is volatile, i.e. whether it can return
     /// different results when evaluated multiple times with the same input.
     /// Note: unlike [`Self::is_volatile`], this function does not consider inputs:

datafusion/expr/src/expr_rewriter/mod.rs

@@ -130,6 +130,28 @@ pub fn normalize_sorts(
         .collect()
 }
 
+/// Recursively rename the table of all [`Column`] expressions in a given expression tree with
+/// a new name, ignoring the `skip_tables`
+pub fn replace_col_base_table(
+    expr: Expr,
+    skip_tables: &[&str],
+    new_table: String,
+) -> Result<Expr> {
+    expr.transform(|expr| {
+        if let Expr::Column(c) = &expr {
+            if let Some(relation) = &c.relation {
+                if !skip_tables.contains(&relation.table()) {
+                    return Ok(Transformed::yes(Expr::Column(
+                        c.with_relation(TableReference::bare(new_table.clone())),
+                    )));
+                }
+            }
+        }
+        Ok(Transformed::no(expr))
+    })
+    .data()
+}
+
 /// Recursively replace all [`Column`] expressions in a given expression tree with
 /// `Column` expressions provided by the hash map argument.
 pub fn replace_col(expr: Expr, replace_map: &HashMap<&Column, &Column>) -> Result<Expr> {

datafusion/expr/src/utils.rs

@@ -933,6 +933,31 @@ pub fn split_conjunction(expr: &Expr) -> Vec<&Expr> {
     split_conjunction_impl(expr, vec![])
 }
 
+/// Splits a conjunctive [`Expr`] such as `A OR B OR C` => `[A, B, C]`
+///
+/// See [`split_disjunction`] for more details and an example.
+pub fn split_disjunction(expr: &Expr) -> Vec<&Expr> {
+    split_disjunction_impl(expr, vec![])
+}
+
+fn split_disjunction_impl<'a>(expr: &'a Expr, mut exprs: Vec<&'a Expr>) -> Vec<&'a Expr> {
+    match expr {
+        Expr::BinaryExpr(BinaryExpr {
+            right,
+            op: Operator::Or,
+            left,
+        }) => {
+            let exprs = split_disjunction_impl(left, exprs);
+            split_disjunction_impl(right, exprs)
+        }
+        Expr::Alias(Alias { expr, .. }) => split_disjunction_impl(expr, exprs),
+        other => {
+            exprs.push(other);
+            exprs
+        }
+    }
+}
+
 fn split_conjunction_impl<'a>(expr: &'a Expr, mut exprs: Vec<&'a Expr>) -> Vec<&'a Expr> {
     match expr {
         Expr::BinaryExpr(BinaryExpr {

datafusion/optimizer/Cargo.toml

@@ -46,6 +46,7 @@ chrono = { workspace = true }
 datafusion-common = { workspace = true, default-features = true }
 datafusion-expr = { workspace = true }
 datafusion-physical-expr = { workspace = true }
+datafusion-sql = { workspace = true }
 indexmap = { workspace = true }
 itertools = { workspace = true }
 log = { workspace = true }