From ad845138beb1a8348e4015367bbe690dda44d31a Mon Sep 17 00:00:00 2001
From: DreaMer963 <wxd963996380@gmail.com>
Date: Fri, 4 Feb 2022 17:19:02 +0800
Subject: [PATCH] finish core

---
 datafusion/src/logical_plan/expr.rs |  2 +-
 datafusion/src/logical_plan/plan.rs | 22 ++++++++++++++++++++--
 2 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/datafusion/src/logical_plan/expr.rs b/datafusion/src/logical_plan/expr.rs
index 2deefa49a9c7..767f9dae355c 100644
--- a/datafusion/src/logical_plan/expr.rs
+++ b/datafusion/src/logical_plan/expr.rs
@@ -557,7 +557,7 @@ impl Expr {
                 // finally, return the column contains `bool`
                 Expr::Column(Column {
                     relation: None,
-                    name: "".to_string(),
+                    name: "exists_subquery_col".to_string(),
                 })
             }
         })
diff --git a/datafusion/src/logical_plan/plan.rs b/datafusion/src/logical_plan/plan.rs
index 0e7c816c6ac2..6e5ce9dfa37e 100644
--- a/datafusion/src/logical_plan/plan.rs
+++ b/datafusion/src/logical_plan/plan.rs
@@ -686,13 +686,31 @@ impl LogicalPlan {
     /// process subqueries
     pub fn process_subquery(self, expr: &LogicalPlan) -> Result<LogicalPlan> {
         let outer_table = self.get_table_scan().unwrap();
-        let distinct_outer_table = LogicalPlanBuilder::from(outer_table)
+        let distinct_outer_table = LogicalPlanBuilder::from(outer_table.clone())
             .distinct()?
             .build()
             .unwrap();
         let result_plan = distinct_outer_table.compute(expr).unwrap();
         // join with outer_table
-        todo!()
+        let join_keys = outer_table
+            .schema()
+            .fields()
+            .iter()
+            .zip(
+                result_plan.schema().fields().as_slice()
+                    [0..outer_table.schema().fields().len()]
+                    .iter(),
+            )
+            .map(|(left_field, right_field)| {
+                (
+                    (Column::from_name(left_field.name())),
+                    (Column::from_name(right_field.name())),
+                )
+            })
+            .unzip();
+        LogicalPlanBuilder::from(outer_table)
+            .join(&result_plan, JoinType::Inner, join_keys)?
+            .build()
     }
 
     /// compute every row in outer_table for subquery