address comments

Author: adrian-wang

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1169,30 +1169,36 @@ class Analyzer(
   object ResolveNaturalJoin extends Rule[LogicalPlan] {
     override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
       // Should not skip unresolved nodes because natural join is always unresolved.
-      case j @ Join(left, right, NaturalJoin(joinType), condition) if j.partlyResolved =>
+      case j @ Join(left, right, NaturalJoin(joinType), condition) if j.resolvedExceptNatural =>
+        // find common column names from both sides, should be treated like usingColumns
         val joinNames = left.output.map(_.name).intersect(right.output.map(_.name))
         val leftKeys = joinNames.map(keyName => left.output.find(_.name == keyName).get)
         val rightKeys = joinNames.map(keyName => right.output.find(_.name == keyName).get)
         val joinPairs = leftKeys.zip(rightKeys)
+        // Add joinPairs to joinConditions
         val newCondition = (condition ++ joinPairs.map {
           case (l, r) => EqualTo(l, r)
         }).reduceLeftOption(And)
+        // columns not in joinPairs
         val lUniqueOutput = left.output.filterNot(att => leftKeys.contains(att))
         val rUniqueOutput = right.output.filterNot(att => rightKeys.contains(att))
+        // we should only keep unique columns(depends on joinType) for joinCols
         val projectList = joinType match {
           case LeftOuter =>
             leftKeys ++ lUniqueOutput ++ rUniqueOutput.map(_.withNullability(true))
           case RightOuter =>
             rightKeys ++ lUniqueOutput.map(_.withNullability(true)) ++ rUniqueOutput
           case FullOuter =>
+            // in full outer join, joinCols should be non-null if there is.
             val joinedCols = joinPairs.map {
               case (l, r) => Alias(Coalesce(Seq(l, r)), l.name)()
             }
             joinedCols ++ lUniqueOutput.map(_.withNullability(true)) ++
               rUniqueOutput.map(_.withNullability(true))
           case _ =>
-            leftKeys ++ lUniqueOutput ++ rUniqueOutput
+            rightKeys ++ lUniqueOutput ++ rUniqueOutput
         }
+        // use Project to trim unnecessary fields
         Project(projectList, Join(left, right, joinType, newCondition))
     }
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -105,9 +105,6 @@ trait CheckAnalysis {
               s"filter expression '${f.condition.prettyString}' " +
                 s"of type ${f.condition.dataType.simpleString} is not a boolean.")
 
-          case j @ Join(_, _, NaturalJoin(_), _) =>
-            failAnalysis(s"natural join not resolved.")
-
           case j @ Join(_, _, _, Some(condition)) if condition.dataType != BooleanType =>
             failAnalysis(
               s"join condition '${condition.prettyString}' " +

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala

@@ -171,16 +171,20 @@ case class Join(
 
   def selfJoinResolved: Boolean = left.outputSet.intersect(right.outputSet).isEmpty
 
-  lazy val partlyResolved: Boolean = {
+  // if not a natural join, it is resolved. if it is a natural join, we still need to
+  // eliminate natural before we mark it resolved, but the node should be ready for
+  // resolution only if everything else is resolved here.
+  lazy val resolvedExceptNatural: Boolean = {
     childrenResolved &&
       expressions.forall(_.resolved) &&
       selfJoinResolved &&
       condition.forall(_.dataType == BooleanType)
   }
+
   // Joins are only resolved if they don't introduce ambiguous expression ids.
   override lazy val resolved: Boolean = joinType match {
     case NaturalJoin(_) => false
-    case _ => partlyResolved
+    case _ => resolvedExceptNatural
   }
 }
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala

@@ -256,6 +256,17 @@ class AnalysisSuite extends AnalysisTest {
     val a = testRelation2.output(0)
     val b = testRelation2.output(1)
     val c = testRelation2.output(2)
+    val testRelation0 = LocalRelation(
+      AttributeReference("a", StringType, nullable = false)(),
+      AttributeReference("b", StringType, nullable = false)(),
+      AttributeReference("c", StringType, nullable = false)())
+    val tt1 = testRelation0.select('a, 'b)
+    val tt2 = testRelation0.select('a, 'c)
+    val aa = testRelation0.output(0)
+    val bb = testRelation0.output(1)
+    val cc = testRelation0.output(2)
+    val truebb = testRelation0.output(1).withNullability(true)
+    val truecc = testRelation0.output(2).withNullability(true)
 
     val plan1 = t1.join(t2, NaturalJoin(Inner), None)
     val expected1 = testRelation2.select(a, b).join(
@@ -273,5 +284,23 @@ class AnalysisSuite extends AnalysisTest {
     val expected4 = testRelation2.select(a, b).join(testRelation2.select(
       a, c), FullOuter, Some(EqualTo(a, a))).select(Alias(Coalesce(Seq(a, a)), "a")(), b, c)
     checkAnalysis(plan4, expected4)
+
+    val plan5 = tt1.join(tt2, NaturalJoin(Inner), None)
+    val expected5 = testRelation0.select(aa, bb).join(
+      testRelation0.select(aa, cc), Inner, Some(EqualTo(aa, aa))).select(aa, bb, cc)
+    checkAnalysis(plan5, expected5)
+    val plan6 = tt1.join(tt2, NaturalJoin(LeftOuter), None)
+    val expected6 = testRelation0.select(aa, bb).join(
+      testRelation0.select(aa, cc), LeftOuter, Some(EqualTo(aa, aa))).select(aa, bb, truecc)
+    checkAnalysis(plan6, expected6)
+    val plan7 = tt1.join(tt2, NaturalJoin(RightOuter), None)
+    val expected7 = testRelation0.select(aa, bb).join(
+      testRelation0.select(aa, cc), RightOuter, Some(EqualTo(aa, aa))).select(aa, truebb, cc)
+    checkAnalysis(plan7, expected7)
+    val plan8 = tt1.join(tt2, NaturalJoin(FullOuter), None)
+    val expected8 = testRelation0.select(aa, bb).join(
+      testRelation0.select(aa, cc), FullOuter, Some(EqualTo(aa, aa))).select(
+      Alias(Coalesce(Seq(aa, aa)), "a")(), truebb, truecc)
+    checkAnalysis(plan8, expected8)
   }
 }