[SPARK-25951][SQL] Remove Alias when canonicalize

Author: mgaido91

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Canonicalize.scala

@@ -34,7 +34,7 @@ package org.apache.spark.sql.catalyst.expressions
  */
 object Canonicalize {
   def execute(e: Expression): Expression = {
-    expressionReorder(ignoreNamesTypes(e))
+    expressionReorder(ignoreNamesTypes(removeAlias(e)))
   }
 
   /** Remove names and nullability from types. */
@@ -91,4 +91,10 @@ object Canonicalize {
 
     case _ => e
   }
+
+  /** Remove Aliases, */
+  private def removeAlias(e: Expression): Expression = e match {
+    case a: Alias => a.child
+    case _ => e
+  }
 }

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

@@ -780,6 +780,23 @@ class PlannerSuite extends SharedSQLContext {
         classOf[PartitioningCollection])
     }
   }
+
+  test("SPARK-25951: avoid redundant shuffle on rename") {
+    val renamedA = Alias(exprA, "a")()
+    val doubleRename = Alias(Alias(exprA, "a")(), "b")()
+    val plan1 = ShuffleExchangeExec(
+      HashPartitioning(exprA :: Nil, 5),
+      DummySparkPlan(outputPartitioning = HashPartitioning(doubleRename :: Nil, 5)))
+    val plan2 = ShuffleExchangeExec(
+      HashPartitioning(exprA :: Nil, 5),
+      DummySparkPlan(outputPartitioning = HashPartitioning(renamedA :: Nil, 5)))
+    val smjExec = SortMergeJoinExec(
+      doubleRename :: Nil, renamedA :: Nil, Inner, None, plan1, plan2)
+
+    val outputPlan = EnsureRequirements(spark.sessionState.conf).apply(smjExec)
+    assertDistributionRequirementsAreSatisfied(outputPlan)
+    assert(outputPlan.collect { case _: ShuffleExchangeExec => true }.isEmpty)
+  }
 }
 
 // Used for unit-testing EnsureRequirements