diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala
index 90711f2b1d..2a8e147011 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala
@@ -79,11 +79,18 @@ case class LogicalRelation(
   /** Used to lookup original attribute capitalization */
   val attributeMap: AttributeMap[AttributeReference] = AttributeMap(output.map(o => (o, o)))
 
-  def newInstance(): this.type =
+  /**
+   * Returns a new instance of this LogicalRelation. According to the semantics of
+   * MultiInstanceRelation, this method returns a copy of this object with
+   * unique expression ids. We respect the `expectedOutputAttributes` and create
+   * new instances of attributes in it.
+   */
+  override def newInstance(): this.type = {
     LogicalRelation(
       relation,
-      expectedOutputAttributes,
+      expectedOutputAttributes.map(_.map(_.newInstance())),
       metastoreTableIdentifier).asInstanceOf[this.type]
+  }
 
   override def refresh(): Unit = relation match {
     case fs: HadoopFsRelation => fs.refresh()
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/parquetSuites.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/parquetSuites.scala
index 31b6197d56..e92bbdea75 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/parquetSuites.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/parquetSuites.scala
@@ -589,6 +589,13 @@ class ParquetMetastoreSuite extends ParquetPartitioningTest {
       }
     }
   }
+
+  test("self-join") {
+    val table = spark.table("normal_parquet")
+    val selfJoin = table.as("t1").join(table.as("t2"))
+    checkAnswer(selfJoin,
+      sql("SELECT * FROM normal_parquet x JOIN normal_parquet y"))
+  }
 }
 
 /**