diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index 37f0c8ed19d3..fa5f1f121818 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++ b/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,8 @@ 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.isInstanceOf[NamedExpression] => b.child
         }
       }
     }.getOrElse(plan)
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
index e607af67f93e..378662fed8bf 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
+++ b/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