Support for SQL Natural Join (apache#4863)

* Support for SQL Natural Join

* Remove comments

Author: Jefffrey

datafusion/sql/src/relation/join.rs

@@ -21,7 +21,7 @@ use datafusion_common::{Column, DFSchemaRef, DataFusionError, Result};
 use datafusion_expr::expr_rewriter::normalize_col_with_schemas;
 use datafusion_expr::{Expr, JoinType, LogicalPlan, LogicalPlanBuilder};
 use sqlparser::ast::{Join, JoinConstraint, JoinOperator, TableWithJoins};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 impl<'a, S: ContextProvider> SqlToRel<'a, S> {
     pub(crate) fn plan_table_with_joins(
@@ -142,10 +142,27 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                     .build()
             }
             JoinConstraint::Natural => {
-                // https://issues.apache.org/jira/browse/ARROW-10727
-                Err(DataFusionError::NotImplemented(
-                    "NATURAL JOIN is not supported (https://issues.apache.org/jira/browse/ARROW-10727)".to_string(),
-                ))
+                let left_cols: HashSet<&String> = left
+                    .schema()
+                    .fields()
+                    .iter()
+                    .map(|f| f.field().name())
+                    .collect();
+                let keys: Vec<Column> = right
+                    .schema()
+                    .fields()
+                    .iter()
+                    .map(|f| f.field().name())
+                    .filter(|f| left_cols.contains(f))
+                    .map(Column::from_name)
+                    .collect();
+                if keys.is_empty() {
+                    self.parse_cross_join(left, right)
+                } else {
+                    LogicalPlanBuilder::from(left)
+                        .join_using(right, join_type, keys)?
+                        .build()
+                }
             }
             JoinConstraint::None => Err(DataFusionError::NotImplemented(
                 "NONE constraint is not supported".to_string(),

datafusion/sql/tests/integration_test.rs

@@ -395,6 +395,54 @@ fn join_with_ambiguous_column() {
     quick_test(sql, expected);
 }
 
+#[test]
+fn natural_join() {
+    let sql = "SELECT * FROM lineitem a NATURAL JOIN lineitem b";
+    let expected = "Projection: a.l_item_id, a.l_description, a.price\
+                        \n  Inner Join: Using a.l_item_id = b.l_item_id, a.l_description = b.l_description, a.price = b.price\
+                        \n    SubqueryAlias: a\
+                        \n      TableScan: lineitem\
+                        \n    SubqueryAlias: b\
+                        \n      TableScan: lineitem";
+    quick_test(sql, expected);
+}
+
+#[test]
+fn natural_left_join() {
+    let sql = "SELECT l_item_id FROM lineitem a NATURAL LEFT JOIN lineitem b";
+    let expected = "Projection: a.l_item_id\
+                        \n  Left Join: Using a.l_item_id = b.l_item_id, a.l_description = b.l_description, a.price = b.price\
+                        \n    SubqueryAlias: a\
+                        \n      TableScan: lineitem\
+                        \n    SubqueryAlias: b\
+                        \n      TableScan: lineitem";
+    quick_test(sql, expected);
+}
+
+#[test]
+fn natural_right_join() {
+    let sql = "SELECT l_item_id FROM lineitem a NATURAL RIGHT JOIN lineitem b";
+    let expected = "Projection: a.l_item_id\
+                        \n  Right Join: Using a.l_item_id = b.l_item_id, a.l_description = b.l_description, a.price = b.price\
+                        \n    SubqueryAlias: a\
+                        \n      TableScan: lineitem\
+                        \n    SubqueryAlias: b\
+                        \n      TableScan: lineitem";
+    quick_test(sql, expected);
+}
+
+#[test]
+fn natural_join_no_common_becomes_cross_join() {
+    let sql = "SELECT * FROM person a NATURAL JOIN lineitem b";
+    let expected = "Projection: a.id, a.first_name, a.last_name, a.age, a.state, a.salary, a.birth_date, a.😀, b.l_item_id, b.l_description, b.price\
+                        \n  CrossJoin:\
+                        \n    SubqueryAlias: a\
+                        \n      TableScan: person\
+                        \n    SubqueryAlias: b\
+                        \n      TableScan: lineitem";
+    quick_test(sql, expected);
+}
+
 #[test]
 fn using_join_multiple_keys() {
     let sql = "SELECT * FROM person a join person b using (id, age)";