Unnest Correlated Subquery

datafusion-cli/Cargo.toml

@@ -31,7 +31,7 @@ rust-version = { workspace = true }
 all-features = true
 
 [features]
-default = []
+default = ["backtrace"]
 backtrace = ["datafusion/backtrace"]
 
 [dependencies]

datafusion/common/src/config.rs

@@ -783,7 +783,7 @@ config_namespace! {
         pub skip_failed_rules: bool, default = false
 
         /// Number of times that the optimizer will attempt to optimize the plan
-        pub max_passes: usize, default = 3
+        pub max_passes: usize, default = 1
 
         /// When set to true, the physical plan optimizer will run a top down
         /// process to reorder the join keys

datafusion/common/src/functional_dependencies.rs

@@ -360,7 +360,7 @@ impl FunctionalDependencies {
                 left_func_dependencies.extend(right_func_dependencies);
                 left_func_dependencies
             }
-            JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark => {
+            JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark | JoinType::LeftSingle => {
                 // These joins preserve functional dependencies of the left side:
                 left_func_dependencies
             }

datafusion/common/src/join_type.rs

@@ -72,6 +72,8 @@ pub enum JoinType {
     /// Same logic as the LeftMark Join above, however it returns a record for each record from the
     /// right input.
     RightMark,
+
+    LeftSingle,
 }
 
 impl JoinType {
@@ -94,6 +96,7 @@ impl JoinType {
             JoinType::RightAnti => JoinType::LeftAnti,
             JoinType::LeftMark => JoinType::RightMark,
             JoinType::RightMark => JoinType::LeftMark,
+            JoinType::LeftSingle => unreachable!(), // TODO: add right single support
         }
     }
 
@@ -126,6 +129,7 @@ impl Display for JoinType {
             JoinType::RightAnti => "RightAnti",
             JoinType::LeftMark => "LeftMark",
             JoinType::RightMark => "RightMark",
+            JoinType::LeftSingle => "LeftSingle",
         };
         write!(f, "{join_type}")
     }
@@ -147,6 +151,7 @@ impl FromStr for JoinType {
             "RIGHTANTI" => Ok(JoinType::RightAnti),
             "LEFTMARK" => Ok(JoinType::LeftMark),
             "RIGHTMARK" => Ok(JoinType::RightMark),
+            "LEFtSINGLE" => Ok(JoinType::LeftSingle),
             _ => _not_impl_err!("The join type {s} does not exist or is not implemented"),
         }
     }

datafusion/core/src/physical_planner.rs

@@ -78,8 +78,8 @@ use datafusion_expr::expr::{
 use datafusion_expr::expr_rewriter::unnormalize_cols;
 use datafusion_expr::logical_plan::builder::wrap_projection_for_join_if_necessary;
 use datafusion_expr::{
-    Analyze, DescribeTable, DmlStatement, Explain, ExplainFormat, Extension, FetchType,
-    Filter, JoinType, RecursiveQuery, SkipType, StringifiedPlan, WindowFrame,
+    Analyze, DelimGet, DescribeTable, DmlStatement, Explain, ExplainFormat, Extension,
+    FetchType, Filter, JoinType, RecursiveQuery, SkipType, StringifiedPlan, WindowFrame,
     WindowFrameBound, WriteOp,
 };
 use datafusion_physical_expr::aggregate::{AggregateExprBuilder, AggregateFunctionExpr};
@@ -1311,6 +1311,63 @@ 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(DelimGet {
+                table_name,
+                projected_schema,
+                ..
+            }) => {
+                let resolved = session_state.resolve_table_ref(table_name.clone());
+                if let Ok(schema) = session_state.schema_for_ref(resolved.clone()) {
+                    if let Some(table) = schema.table(&resolved.table).await? {
+                        let mut proj = vec![];
+                        for (i, field) in table.schema().fields().iter().enumerate() {
+                            for iter in projected_schema.as_ref().iter() {
+                                if iter.1 == field {
+                                    proj.push(i);
+                                }
+                            }
+                        }
+
+                        // First create the scan execution plan.
+                        let scan_plan =
+                            table.scan(session_state, Some(&proj), &[], None).await?;
+
+                        // Now add aggregation to eliminate duplicated rows.
+                        // Create a PhysicalGroupBy with empty expressions, which means we're grouping by all columns
+                        let schema = &scan_plan.schema();
+                        let group_exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = (0
+                            ..schema.fields().len())
+                            .map(|i| {
+                                let name = schema.field(i).name().to_string();
+                                let expr = Arc::new(Column::new(&name, i))
+                                    as Arc<dyn PhysicalExpr>;
+                                (expr, name)
+                            })
+                            .collect();
+
+                        let group_by = PhysicalGroupBy::new_single(group_exprs);
+
+                        // Create the AggregateExec with no aggregate expressions to deduplicate the rows
+                        Arc::new(AggregateExec::try_new(
+                            AggregateMode::Final,
+                            group_by,
+                            vec![], // No aggregate expressions, just grouping to deduplicate
+                            vec![], // No filters
+                            scan_plan.clone(),
+                            scan_plan.schema(),
+                        )?)
+                    } else {
+                        return internal_err!("no table provider");
+                    }
+                } else {
+                    return internal_err!("empty schema");
+                }
+            }
         };
         Ok(exec_node)
     }

datafusion/expr/src/expr.rs

@@ -3169,7 +3169,18 @@ pub const UNNEST_COLUMN_PREFIX: &str = "UNNEST";
 impl Display for Expr {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         match self {
-            Expr::Alias(Alias { expr, name, .. }) => write!(f, "{expr} AS {name}"),
+            Expr::Alias(Alias {
+                expr,
+                relation,
+                name,
+                ..
+            }) => {
+                if let Some(relation) = relation {
+                    write!(f, "{expr} AS {relation}.{name}")
+                } else {
+                    write!(f, "{expr} AS {name}")
+                }
+            }
             Expr::Column(c) => write!(f, "{c}"),
             Expr::OuterReferenceColumn(_, c) => {
                 write!(f, "{OUTER_REFERENCE_COLUMN_PREFIX}({c})")