[SPARK-18609][SQL][WIP]Fix when CTE with Join between two table with same column name

Author: thisissj

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

@@ -26,7 +26,7 @@ import org.apache.spark.api.java.function.FilterFunction
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.{CatalystConf, SimpleCatalystConf}
 import org.apache.spark.sql.catalyst.analysis._
-import org.apache.spark.sql.catalyst.catalog.{InMemoryCatalog, SessionCatalog}
+import org.apache.spark.sql.catalyst.catalog.{CatalogRelation, InMemoryCatalog, SessionCatalog}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral, TrueLiteral}
@@ -200,6 +200,7 @@ object RemoveAliasOnlyProject extends Rule[LogicalPlan] {
         case plan: Project if plan eq proj => plan.child
         case plan => plan transformExpressions {
           case a: Attribute if attrMap.contains(a) => attrMap(a)
+          case b: Alias if (attrMap.exists(_._1.exprId == b.exprId)) => b.child
         }
       }
     }.getOrElse(plan)
@@ -416,8 +417,8 @@ object ColumnPruning extends Rule[LogicalPlan] {
     case w: Window if w.windowExpressions.isEmpty => w.child
 
     // Eliminate no-op Projects
-    case p @ Project(_, child) if sameOutput(child.output, p.output) => child
-
+    case p @ Project(_, child) if sameOutput(child.output, p.output) =>
+      if (child.isInstanceOf[CatalogRelation]) p else child
     // Can't prune the columns on LeafNode
     case p @ Project(_, _: LeafNode) => p
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala

@@ -2011,6 +2011,22 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
     }
   }
 
+  test("test CTE with join between two table with the same column name ") {
+    sql("DROP TABLE IF EXISTS p1")
+    sql("DROP TABLE IF EXISTS p2")
+    sql("CREATE TABLE p1 (col String)" )
+    sql("CREATE TABLE p2 (col String)")
+
+    assert(
+      sql(
+        """
+          | WITH CTE AS
+          |  (SELECT s2.col as col FROM p1
+          |     CROSS JOIN (SELECT e.col as col FROM p2 E) s2)
+          | SELECT T1.col as c1,T2.col as c2 FROM CTE T1 CROSS JOIN CTE T2
+        """.stripMargin).collect.isEmpty)
+  }
+
   def testCommandAvailable(command: String): Boolean = {
     val attempt = Try(Process(command).run(ProcessLogger(_ => ())).exitValue())
     attempt.isSuccess && attempt.get == 0