[WIP] Flatten dependent join

datafusion/core/src/physical_planner.rs

@@ -1246,6 +1246,12 @@ impl DefaultPhysicalPlanner {
                     "Unsupported logical plan: Analyze must be root of the plan"
                 )
             }
+            LogicalPlan::DependentJoin(_) => {
+                return internal_err!(
+                    "Optimizors have not completely remove dependent join"
+                )
+            }
+            LogicalPlan::DelimGet(_) => todo!()
         };
         Ok(exec_node)
     }

datafusion/expr/src/logical_plan/builder.rs

@@ -49,6 +49,7 @@ use crate::{
 
 use super::dml::InsertOp;
 use super::plan::{ColumnUnnestList, ExplainFormat};
+use super::DependentJoin;
 use arrow::compute::can_cast_types;
 use arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef};
 use datafusion_common::display::ToStringifiedPlan;
@@ -717,7 +718,7 @@ impl LogicalPlanBuilder {
                         )
                     })
                     .collect::<Result<Vec<_>>>()?;
-                curr_plan.with_new_exprs(curr_plan.expressions(), new_inputs)
+                curr_plan.with_new_exprs_inputs(curr_plan.expressions(), new_inputs)
             }
         }
     }
@@ -880,6 +881,47 @@ impl LogicalPlanBuilder {
         ))))
     }
 
+    /// Build a dependent join provided a subquery plan
+    /// this function should only be used by the optimizor
+    /// a dependent join node will provides all columns belonging to the LHS
+    /// and one additional column as the result of evaluating the subquery on the RHS
+    /// under the name "subquery_name.output"
+    pub fn dependent_join(
+        self,
+        right: LogicalPlan,
+        correlated_columns: Vec<(usize, Expr)>,
+        subquery_expr: Option<Expr>,
+        subquery_depth: usize,
+        subquery_name: String,
+        lateral_join_condition: Option<(JoinType, Expr)>,
+    ) -> Result<Self> {
+        let left = self.build()?;
+        let schema = left.schema();
+        // TODO: for lateral join, output schema is similar to a normal join
+        let qualified_fields = schema
+            .iter()
+            .map(|(q, f)| (q.cloned(), Arc::clone(f)))
+            .chain(
+                subquery_expr
+                    .iter()
+                    .map(|expr| subquery_output_field(&subquery_name, expr)),
+            )
+            .collect();
+        let metadata = schema.metadata().clone();
+        let dfschema = DFSchema::new_with_metadata(qualified_fields, metadata)?;
+
+        Ok(Self::new(LogicalPlan::DependentJoin(DependentJoin {
+            schema: DFSchemaRef::new(dfschema),
+            left: Arc::new(left),
+            right: Arc::new(right),
+            correlated_columns,
+            subquery_expr,
+            subquery_name,
+            subquery_depth,
+            lateral_join_condition,
+        })))
+    }
+
     /// Apply a join to `right` using explicitly specified columns and an
     /// optional filter expression.
     ///
@@ -1544,6 +1586,26 @@ fn mark_field(schema: &DFSchema) -> (Option<TableReference>, Arc<Field>) {
     )
 }
 
+fn subquery_output_field(
+    subquery_alias: &str,
+    subquery_expr: &Expr,
+) -> (Option<TableReference>, Arc<Field>) {
+    // TODO: check nullability
+    let field = match subquery_expr {
+        Expr::InSubquery(_) => Arc::new(Field::new("output", DataType::Boolean, false)),
+        Expr::Exists(_) => Arc::new(Field::new("output", DataType::Boolean, false)),
+        Expr::ScalarSubquery(sq) => {
+            let data_type = sq.subquery.schema().field(0).data_type().clone();
+            Arc::new(Field::new("output", data_type, false))
+        }
+        _ => {
+            unreachable!()
+        }
+    };
+
+    (Some(TableReference::bare(subquery_alias)), field)
+}
+
 /// Creates a schema for a join operation.
 /// The fields from the left side are first
 pub fn build_join_schema(

datafusion/expr/src/logical_plan/display.rs

@@ -485,6 +485,7 @@ impl<'a, 'b> PgJsonVisitor<'a, 'b> {
 
                 object
             }
+            LogicalPlan::DependentJoin(..) => json!({}),
             LogicalPlan::Join(Join {
                 on: ref keys,
                 filter,
@@ -650,6 +651,7 @@ impl<'a, 'b> PgJsonVisitor<'a, 'b> {
                     "StructColumn": expr_vec_fmt!(struct_type_columns),
                 })
             }
+            LogicalPlan::DelimGet(_) => todo!(),
         }
     }
 }

datafusion/expr/src/logical_plan/mod.rs

@@ -37,11 +37,12 @@ pub use ddl::{
 };
 pub use dml::{DmlStatement, WriteOp};
 pub use plan::{
-    projection_schema, Aggregate, Analyze, ColumnUnnestList, DescribeTable, Distinct,
-    DistinctOn, EmptyRelation, Explain, ExplainFormat, Extension, FetchType, Filter,
-    Join, JoinConstraint, JoinType, Limit, LogicalPlan, Partitioning, PlanType,
-    Projection, RecursiveQuery, Repartition, SkipType, Sort, StringifiedPlan, Subquery,
-    SubqueryAlias, TableScan, ToStringifiedPlan, Union, Unnest, Values, Window,
+    projection_schema, Aggregate, Analyze, ColumnUnnestList, DependentJoin,
+    DescribeTable, Distinct, DistinctOn, EmptyRelation, Explain, ExplainFormat,
+    Extension, FetchType, Filter, Join, JoinConstraint, JoinType, Limit, LogicalPlan,
+    Partitioning, PlanType, Projection, RecursiveQuery, Repartition, SkipType, Sort,
+    StringifiedPlan, Subquery, SubqueryAlias, TableScan, ToStringifiedPlan, Union,
+    Unnest, Values, Window,DelimGet,
 };
 pub use statement::{
     Deallocate, Execute, Prepare, SetVariable, Statement, TransactionAccessMode,