Fix

Author: maropu

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala

@@ -119,7 +119,8 @@ object ReorderJoin extends Rule[LogicalPlan] with PredicateHelper {
         createOrderedJoin(input, conditions)
       }
 
-      // Checks if joins were reordered. If not reordered, returns the original plan
+      // To avoid applying this rule repeatedly, we don't change the plan in case of
+      // the same join order between `p` and `reordered`.
       if (!sameJoinOrder(reordered, p)) {
         if (p.sameOutput(reordered)) {
           reordered
@@ -129,7 +130,7 @@ object ReorderJoin extends Rule[LogicalPlan] with PredicateHelper {
           Project(p.output, reordered)
         }
       } else {
-        reordered
+        p
       }
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala

@@ -250,8 +250,8 @@ object ExtractFiltersAndInnerJoins extends PredicateHelper {
       val (plans, conditions) = flattenJoin(left, joinType)
       (plans ++ Seq((right, joinType)), conditions ++
         cond.toSeq.flatMap(splitConjunctivePredicates))
-    case Filter(filterCondition, j @ Join(_, _, _: InnerLike, _, hint)) if hint == JoinHint.NONE =>
-      val (plans, conditions) = flattenJoin(j)
+    case Filter(filterCondition, child) =>
+      val (plans, conditions) = flattenJoin(child)
       (plans, conditions ++ splitConjunctivePredicates(filterCondition))
     case p @ Project(_, child)
         // Keep flattening joins when the project has attributes only

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinOptimizationSuite.scala

@@ -137,7 +137,7 @@ class JoinOptimizationSuite extends PlanTest {
     }
   }
 
-  test("SPARK-23172 skip projections when flattening joins") {
+  test("Skip projections when flattening joins") {
     def checkExtractInnerJoins(plan: LogicalPlan): Unit = {
       val expectedTables = plan.collectLeaves().map { case p => (p, Inner) }
       val expectedConditions = plan.collect {
@@ -170,7 +170,7 @@ class JoinOptimizationSuite extends PlanTest {
     checkExtractInnerJoins(joined)
   }
 
-  test("SPARK-23172 reorder joins with projections") {
+  test("Reorder joins with projections") {
     withSQLConf(
         SQLConf.STARSCHEMA_DETECTION.key -> "true",
         SQLConf.CBO_ENABLED.key -> "false") {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/StarJoinReorderSuite.scala

@@ -21,7 +21,7 @@ import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.dsl.plans._
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap}
 import org.apache.spark.sql.catalyst.plans.{Inner, PlanTest}
-import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, LocalRelation, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, LocalRelation, LogicalPlan}
 import org.apache.spark.sql.catalyst.rules.RuleExecutor
 import org.apache.spark.sql.catalyst.statsEstimation.{StatsEstimationTestBase, StatsTestPlan}
 import org.apache.spark.sql.internal.SQLConf._
@@ -580,12 +580,7 @@ class StarJoinReorderSuite extends PlanTest with StatsEstimationTestBase {
 
   private def assertEqualPlans(plan1: LogicalPlan, plan2: LogicalPlan): Unit = {
     val analyzed = plan1.analyze
-    val optimized = Optimize.execute(analyzed) match {
-      // `ReorderJoin` adds `Project` to keep the same order of output attributes.
-      // So, we drop a top `Project` for tests.
-      case project: Project => project.child
-      case p => p
-    }
+    val optimized = Optimize.execute(analyzed)
     val expected = plan2.analyze
 
     assert(equivalentOutput(analyzed, expected)) // if this fails, the expected itself is incorrect