From cc47c3eb07a7628faa4277dd87c2837d01c4f175 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Sun, 11 Sep 2016 23:08:17 -0700
Subject: [PATCH 01/10] fix

---
 .../sql/catalyst/catalog/SessionCatalog.scala |  12 +-
 .../spark/sql/execution/command/ddl.scala     |   7 ++
 .../spark/sql/execution/command/tables.scala  |   6 +-
 .../sql/execution/command/DDLSuite.scala      | 104 +++++++++++++++++-
 4 files changed, 115 insertions(+), 14 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index 9fb5db573b70..693e17ab5b24 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -520,11 +520,11 @@ class SessionCatalog(
       tableName: TableIdentifier,
       parts: Seq[CatalogTablePartition],
       ignoreIfExists: Boolean): Unit = {
-    requireExactMatchedPartitionSpec(parts.map(_.spec), getTableMetadata(tableName))
     val db = formatDatabaseName(tableName.database.getOrElse(getCurrentDatabase))
     val table = formatTableName(tableName.table)
     requireDbExists(db)
     requireTableExists(TableIdentifier(table, Option(db)))
+    requireExactMatchedPartitionSpec(parts.map(_.spec), getTableMetadata(tableName))
     externalCatalog.createPartitions(db, table, parts, ignoreIfExists)
   }
 
@@ -537,11 +537,11 @@ class SessionCatalog(
       specs: Seq[TablePartitionSpec],
       ignoreIfNotExists: Boolean,
       purge: Boolean): Unit = {
-    requirePartialMatchedPartitionSpec(specs, getTableMetadata(tableName))
     val db = formatDatabaseName(tableName.database.getOrElse(getCurrentDatabase))
     val table = formatTableName(tableName.table)
     requireDbExists(db)
     requireTableExists(TableIdentifier(table, Option(db)))
+    requirePartialMatchedPartitionSpec(specs, getTableMetadata(tableName))
     externalCatalog.dropPartitions(db, table, specs, ignoreIfNotExists, purge)
   }
 
@@ -556,12 +556,12 @@ class SessionCatalog(
       specs: Seq[TablePartitionSpec],
       newSpecs: Seq[TablePartitionSpec]): Unit = {
     val tableMetadata = getTableMetadata(tableName)
-    requireExactMatchedPartitionSpec(specs, tableMetadata)
-    requireExactMatchedPartitionSpec(newSpecs, tableMetadata)
     val db = formatDatabaseName(tableName.database.getOrElse(getCurrentDatabase))
     val table = formatTableName(tableName.table)
     requireDbExists(db)
     requireTableExists(TableIdentifier(table, Option(db)))
+    requireExactMatchedPartitionSpec(specs, tableMetadata)
+    requireExactMatchedPartitionSpec(newSpecs, tableMetadata)
     externalCatalog.renamePartitions(db, table, specs, newSpecs)
   }
 
@@ -575,11 +575,11 @@ class SessionCatalog(
    * this becomes a no-op.
    */
   def alterPartitions(tableName: TableIdentifier, parts: Seq[CatalogTablePartition]): Unit = {
-    requireExactMatchedPartitionSpec(parts.map(_.spec), getTableMetadata(tableName))
     val db = formatDatabaseName(tableName.database.getOrElse(getCurrentDatabase))
     val table = formatTableName(tableName.table)
     requireDbExists(db)
     requireTableExists(TableIdentifier(table, Option(db)))
+    requireExactMatchedPartitionSpec(parts.map(_.spec), getTableMetadata(tableName))
     externalCatalog.alterPartitions(db, table, parts)
   }
 
@@ -588,11 +588,11 @@ class SessionCatalog(
    * If no database is specified, assume the table is in the current database.
    */
   def getPartition(tableName: TableIdentifier, spec: TablePartitionSpec): CatalogTablePartition = {
-    requireExactMatchedPartitionSpec(Seq(spec), getTableMetadata(tableName))
     val db = formatDatabaseName(tableName.database.getOrElse(getCurrentDatabase))
     val table = formatTableName(tableName.table)
     requireDbExists(db)
     requireTableExists(TableIdentifier(table, Option(db)))
+    requireExactMatchedPartitionSpec(Seq(spec), getTableMetadata(tableName))
     externalCatalog.getPartition(db, table, spec)
   }
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index bc1c4f85e331..3e9548ae4c4c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -28,6 +28,7 @@ import org.apache.hadoop.mapred.{FileInputFormat, JobConf}
 
 import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
 import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable, CatalogTablePartition, CatalogTableType, SessionCatalog}
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
@@ -266,6 +267,12 @@ case class AlterTableUnsetPropertiesCommand(
     val catalog = sparkSession.sessionState.catalog
     DDLUtils.verifyAlterTableType(catalog, tableName, isView)
     val table = catalog.getTableMetadata(tableName)
+
+    if (catalog.isTemporaryTable(tableName)) {
+      throw new NoSuchTableException(
+        db = tableName.database.getOrElse(catalog.getCurrentDatabase),
+        table = tableName.table)
+    }
     if (!ifExists) {
       propKeys.foreach { k =>
         if (!table.properties.contains(k)) {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 027f3588e292..742d3cd3dc8d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -219,7 +219,7 @@ case class LoadDataCommand(
       throw new AnalysisException(s"Target table in LOAD DATA does not exist: $table")
     }
     val targetTable = catalog.getTableMetadataOption(table).getOrElse {
-      throw new AnalysisException(s"Target table in LOAD DATA cannot be temporary: $table")
+      throw new AnalysisException(s"Operation not allowed: LOAD DATA on temporary tables: $table")
     }
     if (targetTable.tableType == CatalogTableType.VIEW) {
       throw new AnalysisException(s"Target table in LOAD DATA cannot be a view: $table")
@@ -644,7 +644,7 @@ case class ShowPartitionsCommand(
 
     if (catalog.isTemporaryTable(table)) {
       throw new AnalysisException(
-        s"SHOW PARTITIONS is not allowed on a temporary table: ${table.unquotedString}")
+        s"Operation not allowed: SHOW PARTITIONS on temporary tables: $table")
     }
 
     val tab = catalog.getTableMetadata(table)
@@ -702,7 +702,7 @@ case class ShowCreateTableCommand(table: TableIdentifier) extends RunnableComman
 
     if (catalog.isTemporaryTable(table)) {
       throw new AnalysisException(
-        s"SHOW CREATE TABLE cannot be applied to temporary table")
+        s"Operation not allowed: SHOW CREATE TABLE on temporary tables: $table")
     }
 
     if (!catalog.tableExists(table)) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index 05f826a11b58..9a2482bd47c5 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -729,6 +729,100 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
     }
   }
 
+  test("Issue exceptions for ALTER VIEW on the temporary view") {
+    val sourceViewName = "testView"
+    withTempView(sourceViewName) {
+      createTempView(sourceViewName)
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER VIEW $sourceViewName SET TBLPROPERTIES ('p' = 'an')")
+      }
+      intercept[NoSuchTableException] {
+        sql(s"ALTER VIEW $sourceViewName UNSET TBLPROPERTIES ('p')")
+      }
+    }
+  }
+
+  test("Issue exceptions for ALTER TABLE on the temporary view") {
+    val sourceViewName = "testView"
+    withTempView(sourceViewName) {
+      createTempView(sourceViewName)
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $sourceViewName SET SERDE 'whatever'")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $sourceViewName PARTITION (a=1, b=2) SET SERDE 'whatever'")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $sourceViewName SET SERDEPROPERTIES ('p' = 'an')")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $sourceViewName SET LOCATION '/path/to/your/lovely/heart'")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $sourceViewName PARTITION (a='4') SET LOCATION '/path/to/home'")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $sourceViewName ADD IF NOT EXISTS PARTITION (a='4', b='8')")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $sourceViewName DROP PARTITION (a='4', b='8')")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $sourceViewName PARTITION (a='4') RENAME TO PARTITION (a='5')")
+      }
+
+      val e = intercept[AnalysisException] {
+        sql(s"ALTER TABLE $sourceViewName RECOVER PARTITIONS")
+      }.getMessage
+      assert(e.contains(
+        "Operation not allowed: ALTER TABLE RECOVER PARTITIONS on temporary tables: `testView`"))
+    }
+  }
+
+  test("Issue exceptions for other table DDL on the temporary view") {
+    val sourceViewName = "testView"
+    withTempView(sourceViewName) {
+      createTempView(sourceViewName)
+
+      var e = intercept[AnalysisException] {
+        sql(s"TRUNCATE TABLE $sourceViewName")
+      }.getMessage
+      assert(e.contains("Operation not allowed: TRUNCATE TABLE on temporary tables: `testView`"))
+
+      e = intercept[AnalysisException] {
+        sql(s"LOAD DATA LOCAL INPATH '/path/to/home' INTO TABLE $sourceViewName")
+      }.getMessage
+      assert(e.contains("Operation not allowed: LOAD DATA on temporary tables: `testView`"))
+
+      e = intercept[AnalysisException] {
+        sql(s"SHOW CREATE TABLE $sourceViewName")
+      }.getMessage
+      assert(e.contains("Operation not allowed: SHOW CREATE TABLE on temporary tables: `testView`"))
+
+      e = intercept[AnalysisException] {
+        sql(s"SHOW PARTITIONS $sourceViewName")
+      }.getMessage
+      assert(e.contains("Operation not allowed: SHOW PARTITIONS on temporary tables: `testView`"))
+    }
+  }
+
+  private def createTempView(viewName: String): Unit = {
+    sql(
+      s"""
+         |CREATE TEMPORARY VIEW $viewName
+         |AS SELECT 1, 2, 3
+       """.stripMargin)
+  }
+
   test("alter table: set location") {
     testSetLocation(isDatasourceTable = false)
   }
@@ -911,7 +1005,7 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
     withTempView("show1a", "show2b") {
       sql(
         """
-          |CREATE TEMPORARY TABLE show1a
+          |CREATE TEMPORARY VIEW show1a
           |USING org.apache.spark.sql.sources.DDLScanSource
           |OPTIONS (
           |  From '1',
@@ -922,7 +1016,7 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
         """.stripMargin)
       sql(
         """
-          |CREATE TEMPORARY TABLE show2b
+          |CREATE TEMPORARY VIEW show2b
           |USING org.apache.spark.sql.sources.DDLScanSource
           |OPTIONS (
           |  From '1',
@@ -976,11 +1070,11 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
       Nil)
   }
 
-  test("drop table - temporary table") {
+  test("drop table - temporary view") {
     val catalog = spark.sessionState.catalog
     sql(
       """
-        |CREATE TEMPORARY TABLE tab1
+        |CREATE TEMPORARY VIEW tab1
         |USING org.apache.spark.sql.sources.DDLScanSource
         |OPTIONS (
         |  From '1',
@@ -1603,7 +1697,7 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
     }
   }
 
-  test("truncate table - external table, temporary table, view (not allowed)") {
+  test("truncate table - external table, temporary view, view (not allowed)") {
     import testImplicits._
     val path = Utils.createTempDir().getAbsolutePath
     (1 to 10).map { i => (i, i) }.toDF("a", "b").createTempView("my_temp_tab")

From 855df616b958c2b42a3fab8389e94a416ac90f11 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Mon, 12 Sep 2016 21:12:47 -0700
Subject: [PATCH 02/10] added more test cases

---
 .../command/AnalyzeTableCommand.scala         |  5 +-
 .../sql/execution/command/DDLSuite.scala      | 94 -------------------
 .../sql/hive/execution/HiveCommandSuite.scala |  2 +-
 .../sql/hive/execution/SQLViewSuite.scala     | 90 +++++++++++++++++-
 4 files changed, 92 insertions(+), 99 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala
index 15687ddd728a..9f0e02b40ed2 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala
@@ -24,7 +24,7 @@ import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.spark.sql.{AnalysisException, Dataset, Row, SparkSession}
 import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
 import org.apache.spark.sql.catalyst.catalog.{CatalogRelation, CatalogTable}
-import org.apache.spark.sql.catalyst.plans.logical.Statistics
+import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, Statistics}
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 
 
@@ -91,6 +91,9 @@ case class AnalyzeTableCommand(tableName: String, noscan: Boolean = true) extend
       case logicalRel: LogicalRelation if logicalRel.catalogTable.isDefined =>
         updateTableStats(logicalRel.catalogTable.get, logicalRel.relation.sizeInBytes)
 
+      case o if !o.isInstanceOf[LeafNode] =>
+        throw new AnalysisException(s"Operation not allowed: ANALYZE TABLE on views: $tableName")
+
       case otherRelation =>
         throw new AnalysisException(s"ANALYZE TABLE is not supported for " +
           s"${otherRelation.nodeName}.")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index 9a2482bd47c5..2374792a2ab7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -729,100 +729,6 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
     }
   }
 
-  test("Issue exceptions for ALTER VIEW on the temporary view") {
-    val sourceViewName = "testView"
-    withTempView(sourceViewName) {
-      createTempView(sourceViewName)
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER VIEW $sourceViewName SET TBLPROPERTIES ('p' = 'an')")
-      }
-      intercept[NoSuchTableException] {
-        sql(s"ALTER VIEW $sourceViewName UNSET TBLPROPERTIES ('p')")
-      }
-    }
-  }
-
-  test("Issue exceptions for ALTER TABLE on the temporary view") {
-    val sourceViewName = "testView"
-    withTempView(sourceViewName) {
-      createTempView(sourceViewName)
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $sourceViewName SET SERDE 'whatever'")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $sourceViewName PARTITION (a=1, b=2) SET SERDE 'whatever'")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $sourceViewName SET SERDEPROPERTIES ('p' = 'an')")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $sourceViewName SET LOCATION '/path/to/your/lovely/heart'")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $sourceViewName PARTITION (a='4') SET LOCATION '/path/to/home'")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $sourceViewName ADD IF NOT EXISTS PARTITION (a='4', b='8')")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $sourceViewName DROP PARTITION (a='4', b='8')")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $sourceViewName PARTITION (a='4') RENAME TO PARTITION (a='5')")
-      }
-
-      val e = intercept[AnalysisException] {
-        sql(s"ALTER TABLE $sourceViewName RECOVER PARTITIONS")
-      }.getMessage
-      assert(e.contains(
-        "Operation not allowed: ALTER TABLE RECOVER PARTITIONS on temporary tables: `testView`"))
-    }
-  }
-
-  test("Issue exceptions for other table DDL on the temporary view") {
-    val sourceViewName = "testView"
-    withTempView(sourceViewName) {
-      createTempView(sourceViewName)
-
-      var e = intercept[AnalysisException] {
-        sql(s"TRUNCATE TABLE $sourceViewName")
-      }.getMessage
-      assert(e.contains("Operation not allowed: TRUNCATE TABLE on temporary tables: `testView`"))
-
-      e = intercept[AnalysisException] {
-        sql(s"LOAD DATA LOCAL INPATH '/path/to/home' INTO TABLE $sourceViewName")
-      }.getMessage
-      assert(e.contains("Operation not allowed: LOAD DATA on temporary tables: `testView`"))
-
-      e = intercept[AnalysisException] {
-        sql(s"SHOW CREATE TABLE $sourceViewName")
-      }.getMessage
-      assert(e.contains("Operation not allowed: SHOW CREATE TABLE on temporary tables: `testView`"))
-
-      e = intercept[AnalysisException] {
-        sql(s"SHOW PARTITIONS $sourceViewName")
-      }.getMessage
-      assert(e.contains("Operation not allowed: SHOW PARTITIONS on temporary tables: `testView`"))
-    }
-  }
-
-  private def createTempView(viewName: String): Unit = {
-    sql(
-      s"""
-         |CREATE TEMPORARY VIEW $viewName
-         |AS SELECT 1, 2, 3
-       """.stripMargin)
-  }
-
   test("alter table: set location") {
     testSetLocation(isDatasourceTable = false)
   }
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala
index df33731df2d0..5db15cff16bc 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala
@@ -409,7 +409,7 @@ class HiveCommandSuite extends QueryTest with SQLTestUtils with TestHiveSingleto
       val message1 = intercept[AnalysisException] {
         sql("SHOW PARTITIONS parquet_temp")
       }.getMessage
-      assert(message1.contains("is not allowed on a temporary table"))
+      assert(message1.contains("Operation not allowed: SHOW PARTITIONS on temporary tables"))
 
       val message2 = intercept[AnalysisException] {
         sql("SHOW PARTITIONS parquet_tab3")
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
index bc999d472406..f4aafc964f4c 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
@@ -82,10 +82,70 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
     }
   }
 
-  test("error handling: insert/load/truncate table commands against a temp view") {
+  test("Issue exceptions for ALTER VIEW on the temporary view") {
     val viewName = "testView"
     withTempView(viewName) {
-      sql(s"CREATE TEMPORARY VIEW $viewName AS SELECT id FROM jt")
+      createTempView(viewName)
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER VIEW $viewName SET TBLPROPERTIES ('p' = 'an')")
+      }
+      intercept[NoSuchTableException] {
+        sql(s"ALTER VIEW $viewName UNSET TBLPROPERTIES ('p')")
+      }
+    }
+  }
+
+  test("Issue exceptions for ALTER TABLE on the temporary view") {
+    val viewName = "testView"
+    withTempView(viewName) {
+      createTempView(viewName)
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $viewName SET SERDE 'whatever'")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $viewName PARTITION (a=1, b=2) SET SERDE 'whatever'")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $viewName SET SERDEPROPERTIES ('p' = 'an')")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $viewName SET LOCATION '/path/to/your/lovely/heart'")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $viewName PARTITION (a='4') SET LOCATION '/path/to/home'")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $viewName ADD IF NOT EXISTS PARTITION (a='4', b='8')")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $viewName DROP PARTITION (a='4', b='8')")
+      }
+
+      intercept[NoSuchTableException] {
+        sql(s"ALTER TABLE $viewName PARTITION (a='4') RENAME TO PARTITION (a='5')")
+      }
+
+      val e = intercept[AnalysisException] {
+        sql(s"ALTER TABLE $viewName RECOVER PARTITIONS")
+      }.getMessage
+      assert(e.contains(
+        "Operation not allowed: ALTER TABLE RECOVER PARTITIONS on temporary tables: `testView`"))
+    }
+  }
+
+  test("Issue exceptions for other table DDL on the temporary view") {
+    val viewName = "testView"
+    withTempView(viewName) {
+      createTempView(viewName)
+
       var e = intercept[AnalysisException] {
         sql(s"INSERT INTO TABLE $viewName SELECT 1")
       }.getMessage
@@ -95,15 +155,39 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       e = intercept[AnalysisException] {
         sql(s"""LOAD DATA LOCAL INPATH "$testData" INTO TABLE $viewName""")
       }.getMessage
-      assert(e.contains(s"Target table in LOAD DATA cannot be temporary: `$viewName`"))
+      assert(e.contains(s"Operation not allowed: LOAD DATA on temporary tables: `$viewName`"))
 
       e = intercept[AnalysisException] {
         sql(s"TRUNCATE TABLE $viewName")
       }.getMessage
       assert(e.contains(s"Operation not allowed: TRUNCATE TABLE on temporary tables: `$viewName`"))
+
+      e = intercept[AnalysisException] {
+        sql(s"SHOW CREATE TABLE $viewName")
+      }.getMessage
+      assert(e.contains(
+        s"Operation not allowed: SHOW CREATE TABLE on temporary tables: `$viewName`"))
+
+      e = intercept[AnalysisException] {
+        sql(s"SHOW PARTITIONS $viewName")
+      }.getMessage
+      assert(e.contains(s"Operation not allowed: SHOW PARTITIONS on temporary tables: `$viewName`"))
+
+      e = intercept[AnalysisException] {
+        sql(s"ANALYZE TABLE $viewName COMPUTE STATISTICS")
+      }.getMessage
+      assert(e.contains(s"Operation not allowed: ANALYZE TABLE on views: `$viewName`"))
     }
   }
 
+  private def createTempView(viewName: String): Unit = {
+    sql(
+      s"""
+         |CREATE TEMPORARY VIEW $viewName
+         |AS SELECT 1, 2, 3
+       """.stripMargin)
+  }
+
   test("error handling: insert/load/truncate table commands against a view") {
     val viewName = "testView"
     withView(viewName) {

From 60371d85ee72dc1a6d945e7485c5637e064ac068 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Wed, 14 Sep 2016 10:11:49 -0700
Subject: [PATCH 03/10] address comments

---
 .../sql/execution/command/AnalyzeTableCommand.scala    | 10 +++++-----
 .../org/apache/spark/sql/execution/command/ddl.scala   | 10 +++-------
 .../apache/spark/sql/hive/execution/SQLViewSuite.scala |  5 ++---
 3 files changed, 10 insertions(+), 15 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala
index 9f0e02b40ed2..ad27e0a53f35 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala
@@ -22,9 +22,10 @@ import scala.util.control.NonFatal
 import org.apache.hadoop.fs.{FileSystem, Path}
 
 import org.apache.spark.sql.{AnalysisException, Dataset, Row, SparkSession}
+import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
 import org.apache.spark.sql.catalyst.catalog.{CatalogRelation, CatalogTable}
-import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, Statistics}
+import org.apache.spark.sql.catalyst.plans.logical.Statistics
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 
 
@@ -37,7 +38,9 @@ case class AnalyzeTableCommand(tableName: String, noscan: Boolean = true) extend
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val sessionState = sparkSession.sessionState
     val tableIdent = sessionState.sqlParser.parseTableIdentifier(tableName)
-    val relation = EliminateSubqueryAliases(sessionState.catalog.lookupRelation(tableIdent))
+    val db = tableIdent.database.getOrElse(sessionState.catalog.getCurrentDatabase)
+    val qualifiedName = TableIdentifier(tableIdent.table, Some(db))
+    val relation = EliminateSubqueryAliases(sessionState.catalog.lookupRelation(qualifiedName))
 
     relation match {
       case relation: CatalogRelation =>
@@ -91,9 +94,6 @@ case class AnalyzeTableCommand(tableName: String, noscan: Boolean = true) extend
       case logicalRel: LogicalRelation if logicalRel.catalogTable.isDefined =>
         updateTableStats(logicalRel.catalogTable.get, logicalRel.relation.sizeInBytes)
 
-      case o if !o.isInstanceOf[LeafNode] =>
-        throw new AnalysisException(s"Operation not allowed: ANALYZE TABLE on views: $tableName")
-
       case otherRelation =>
         throw new AnalysisException(s"ANALYZE TABLE is not supported for " +
           s"${otherRelation.nodeName}.")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index 3e9548ae4c4c..850c840fdad3 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -28,7 +28,6 @@ import org.apache.hadoop.mapred.{FileInputFormat, JobConf}
 
 import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
 import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable, CatalogTablePartition, CatalogTableType, SessionCatalog}
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
@@ -266,13 +265,10 @@ case class AlterTableUnsetPropertiesCommand(
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
     DDLUtils.verifyAlterTableType(catalog, tableName, isView)
-    val table = catalog.getTableMetadata(tableName)
+    val db = tableName.database.getOrElse(catalog.getCurrentDatabase)
+    val qualifiedName = TableIdentifier(tableName.table, Some(db))
+    val table = catalog.getTableMetadata(qualifiedName)
 
-    if (catalog.isTemporaryTable(tableName)) {
-      throw new NoSuchTableException(
-        db = tableName.database.getOrElse(catalog.getCurrentDatabase),
-        table = tableName.table)
-    }
     if (!ifExists) {
       propKeys.foreach { k =>
         if (!table.properties.contains(k)) {
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
index f4aafc964f4c..0b629180d3c5 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
@@ -173,10 +173,9 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       }.getMessage
       assert(e.contains(s"Operation not allowed: SHOW PARTITIONS on temporary tables: `$viewName`"))
 
-      e = intercept[AnalysisException] {
+      intercept[NoSuchTableException] {
         sql(s"ANALYZE TABLE $viewName COMPUTE STATISTICS")
-      }.getMessage
-      assert(e.contains(s"Operation not allowed: ANALYZE TABLE on views: `$viewName`"))
+      }
     }
   }
 

From 6029a95f2245d55a00fefa9a99ecd12a5460ffd4 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Wed, 14 Sep 2016 11:12:43 -0700
Subject: [PATCH 04/10] address comments

---
 .../sql/catalyst/analysis/CheckAnalysis.scala |   1 +
 .../spark/sql/execution/command/ddl.scala     |  17 +--
 .../spark/sql/execution/command/tables.scala  |  76 +++++-------
 .../sql/execution/command/DDLSuite.scala      |  14 ++-
 .../sql/hive/execution/SQLViewSuite.scala     | 116 +++++-------------
 5 files changed, 72 insertions(+), 152 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
index e07e9194bee9..9c06069f24f7 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
@@ -360,6 +360,7 @@ trait CheckAnalysis extends PredicateHelper {
 
           case InsertIntoTable(t, _, _, _, _)
             if !t.isInstanceOf[LeafNode] ||
+              t.isInstanceOf[Range] ||
               t == OneRowRelation ||
               t.isInstanceOf[LocalRelation] =>
             failAnalysis(s"Inserting into an RDD-based table is not allowed.")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index 850c840fdad3..8fee2a523389 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -470,25 +470,20 @@ case class AlterTableRecoverPartitionsCommand(
 
   override def run(spark: SparkSession): Seq[Row] = {
     val catalog = spark.sessionState.catalog
-    if (!catalog.tableExists(tableName)) {
-      throw new AnalysisException(s"Table $tableName in $cmd does not exist.")
-    }
-    if (catalog.isTemporaryTable(tableName)) {
-      throw new AnalysisException(
-        s"Operation not allowed: $cmd on temporary tables: $tableName")
-    }
-    val table = catalog.getTableMetadata(tableName)
+    val db = tableName.database.getOrElse(catalog.getCurrentDatabase)
+    val qualifiedName = TableIdentifier(tableName.table, Some(db))
+    val table = catalog.getTableMetadata(qualifiedName)
     if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
-        s"Operation not allowed: $cmd on datasource tables: $tableName")
+        s"Operation not allowed: $cmd on datasource tables: $qualifiedName")
     }
     if (table.partitionColumnNames.isEmpty) {
       throw new AnalysisException(
-        s"Operation not allowed: $cmd only works on partitioned tables: $tableName")
+        s"Operation not allowed: $cmd only works on partitioned tables: $qualifiedName")
     }
     if (table.storage.locationUri.isEmpty) {
       throw new AnalysisException(
-        s"Operation not allowed: $cmd only works on table with location provided: $tableName")
+        s"Operation not allowed: $cmd only works on table with location provided: $qualifiedName")
     }
 
     val root = new Path(table.storage.locationUri.get)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index f44691fafda5..5c599f003a01 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -35,7 +35,6 @@ import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
 import org.apache.spark.sql.catalyst.util.quoteIdentifier
 import org.apache.spark.sql.execution.datasources.PartitioningUtils
-import org.apache.spark.sql.internal.HiveSerDe
 import org.apache.spark.sql.types._
 import org.apache.spark.util.Utils
 
@@ -215,39 +214,38 @@ case class LoadDataCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    if (!catalog.tableExists(table)) {
-      throw new AnalysisException(s"Target table in LOAD DATA does not exist: $table")
-    }
-    val targetTable = catalog.getTableMetadataOption(table).getOrElse {
-      throw new AnalysisException(s"Operation not allowed: LOAD DATA on temporary tables: $table")
-    }
+    val db = table.database.getOrElse(catalog.getCurrentDatabase)
+    val qualifiedName = TableIdentifier(table.table, Some(db))
+    val targetTable = catalog.getTableMetadata(qualifiedName)
+
     if (targetTable.tableType == CatalogTableType.VIEW) {
-      throw new AnalysisException(s"Target table in LOAD DATA cannot be a view: $table")
+      throw new AnalysisException(s"Target table in LOAD DATA cannot be a view: $qualifiedName")
     }
     if (DDLUtils.isDatasourceTable(targetTable)) {
-      throw new AnalysisException(s"LOAD DATA is not supported for datasource tables: $table")
+      throw new AnalysisException(
+        s"LOAD DATA is not supported for datasource tables: $qualifiedName")
     }
     if (targetTable.partitionColumnNames.nonEmpty) {
       if (partition.isEmpty) {
-        throw new AnalysisException(s"LOAD DATA target table $table is partitioned, " +
+        throw new AnalysisException(s"LOAD DATA target table $qualifiedName is partitioned, " +
           s"but no partition spec is provided")
       }
       if (targetTable.partitionColumnNames.size != partition.get.size) {
-        throw new AnalysisException(s"LOAD DATA target table $table is partitioned, " +
+        throw new AnalysisException(s"LOAD DATA target table $qualifiedName is partitioned, " +
           s"but number of columns in provided partition spec (${partition.get.size}) " +
           s"do not match number of partitioned columns in table " +
           s"(s${targetTable.partitionColumnNames.size})")
       }
       partition.get.keys.foreach { colName =>
         if (!targetTable.partitionColumnNames.contains(colName)) {
-          throw new AnalysisException(s"LOAD DATA target table $table is partitioned, " +
+          throw new AnalysisException(s"LOAD DATA target table $qualifiedName is partitioned, " +
             s"but the specified partition spec refers to a column that is not partitioned: " +
             s"'$colName'")
         }
       }
     } else {
       if (partition.nonEmpty) {
-        throw new AnalysisException(s"LOAD DATA target table $table is not partitioned, " +
+        throw new AnalysisException(s"LOAD DATA target table $qualifiedName is not partitioned, " +
           s"but a partition spec was provided.")
       }
     }
@@ -336,32 +334,28 @@ case class TruncateTableCommand(
 
   override def run(spark: SparkSession): Seq[Row] = {
     val catalog = spark.sessionState.catalog
-    if (!catalog.tableExists(tableName)) {
-      throw new AnalysisException(s"Table $tableName in TRUNCATE TABLE does not exist.")
-    }
-    if (catalog.isTemporaryTable(tableName)) {
-      throw new AnalysisException(
-        s"Operation not allowed: TRUNCATE TABLE on temporary tables: $tableName")
-    }
-    val table = catalog.getTableMetadata(tableName)
+    val db = tableName.database.getOrElse(catalog.getCurrentDatabase)
+    val qualifiedName = TableIdentifier(tableName.table, Some(db))
+    val table = catalog.getTableMetadata(qualifiedName)
+
     if (table.tableType == CatalogTableType.EXTERNAL) {
       throw new AnalysisException(
-        s"Operation not allowed: TRUNCATE TABLE on external tables: $tableName")
+        s"Operation not allowed: TRUNCATE TABLE on external tables: $qualifiedName")
     }
     if (table.tableType == CatalogTableType.VIEW) {
       throw new AnalysisException(
-        s"Operation not allowed: TRUNCATE TABLE on views: $tableName")
+        s"Operation not allowed: TRUNCATE TABLE on views: $qualifiedName")
     }
     val isDatasourceTable = DDLUtils.isDatasourceTable(table)
     if (isDatasourceTable && partitionSpec.isDefined) {
       throw new AnalysisException(
         s"Operation not allowed: TRUNCATE TABLE ... PARTITION is not supported " +
-        s"for tables created using the data sources API: $tableName")
+        s"for tables created using the data sources API: $qualifiedName")
     }
     if (table.partitionColumnNames.isEmpty && partitionSpec.isDefined) {
       throw new AnalysisException(
         s"Operation not allowed: TRUNCATE TABLE ... PARTITION is not supported " +
-        s"for tables that are not partitioned: $tableName")
+        s"for tables that are not partitioned: $qualifiedName")
     }
     val locations =
       if (isDatasourceTable) {
@@ -369,7 +363,7 @@ case class TruncateTableCommand(
       } else if (table.partitionColumnNames.isEmpty) {
         Seq(table.storage.locationUri)
       } else {
-        catalog.listPartitions(tableName, partitionSpec).map(_.storage.locationUri)
+        catalog.listPartitions(qualifiedName, partitionSpec).map(_.storage.locationUri)
       }
     val hadoopConf = spark.sessionState.newHadoopConf()
     locations.foreach { location =>
@@ -382,7 +376,7 @@ case class TruncateTableCommand(
         } catch {
           case NonFatal(e) =>
             throw new AnalysisException(
-              s"Failed to truncate table $tableName when removing data of the path: $path " +
+              s"Failed to truncate table $qualifiedName when removing data of the path: $path " +
                 s"because of ${e.toString}")
         }
       }
@@ -392,10 +386,10 @@ case class TruncateTableCommand(
     spark.sessionState.refreshTable(tableName.unquotedString)
     // Also try to drop the contents of the table from the columnar cache
     try {
-      spark.sharedState.cacheManager.uncacheQuery(spark.table(tableName.quotedString))
+      spark.sharedState.cacheManager.uncacheQuery(spark.table(qualifiedName.quotedString))
     } catch {
       case NonFatal(e) =>
-        log.warn(s"Exception when attempting to uncache table $tableName", e)
+        log.warn(s"Exception when attempting to uncache table $qualifiedName", e)
     }
     Seq.empty[Row]
   }
@@ -642,13 +636,9 @@ case class ShowPartitionsCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-
-    if (catalog.isTemporaryTable(table)) {
-      throw new AnalysisException(
-        s"Operation not allowed: SHOW PARTITIONS on temporary tables: $table")
-    }
-
-    val tab = catalog.getTableMetadata(table)
+    val db = table.database.getOrElse(catalog.getCurrentDatabase)
+    val qualifiedName = TableIdentifier(table.table, Some(db))
+    val tab = catalog.getTableMetadata(qualifiedName)
 
     /**
      * Validate and throws an [[AnalysisException]] exception under the following conditions:
@@ -700,17 +690,9 @@ case class ShowCreateTableCommand(table: TableIdentifier) extends RunnableComman
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-
-    if (catalog.isTemporaryTable(table)) {
-      throw new AnalysisException(
-        s"Operation not allowed: SHOW CREATE TABLE on temporary tables: $table")
-    }
-
-    if (!catalog.tableExists(table)) {
-      throw new AnalysisException(s"Table $table doesn't exist")
-    }
-
-    val tableMetadata = catalog.getTableMetadata(table)
+    val db = table.database.getOrElse(catalog.getCurrentDatabase)
+    val qualifiedName = TableIdentifier(table.table, Some(db))
+    val tableMetadata = catalog.getTableMetadata(qualifiedName)
 
     // TODO: unify this after we unify the CREATE TABLE syntax for hive serde and data source table.
     val stmt = if (DDLUtils.isDatasourceTable(tableMetadata)) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index f24ae4a24e3e..83db318ac9b9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -913,7 +913,7 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
     withTempView("show1a", "show2b") {
       sql(
         """
-          |CREATE TEMPORARY VIEW show1a
+          |CREATE TEMPORARY TABLE show1a
           |USING org.apache.spark.sql.sources.DDLScanSource
           |OPTIONS (
           |  From '1',
@@ -924,7 +924,7 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
         """.stripMargin)
       sql(
         """
-          |CREATE TEMPORARY VIEW show2b
+          |CREATE TEMPORARY TABLE show2b
           |USING org.apache.spark.sql.sources.DDLScanSource
           |OPTIONS (
           |  From '1',
@@ -978,11 +978,11 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
       Nil)
   }
 
-  test("drop table - temporary view") {
+  test("drop table - temporary table") {
     val catalog = spark.sessionState.catalog
     sql(
       """
-        |CREATE TEMPORARY VIEW tab1
+        |CREATE TEMPORARY TABLE tab1
         |USING org.apache.spark.sql.sources.DDLScanSource
         |OPTIONS (
         |  From '1',
@@ -1605,13 +1605,15 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
     }
   }
 
-  test("truncate table - external table, temporary view, view (not allowed)") {
+  test("truncate table - external table, temporary table, view (not allowed)") {
     import testImplicits._
     val path = Utils.createTempDir().getAbsolutePath
     (1 to 10).map { i => (i, i) }.toDF("a", "b").createTempView("my_temp_tab")
     sql(s"CREATE EXTERNAL TABLE my_ext_tab LOCATION '$path'")
     sql(s"CREATE VIEW my_view AS SELECT 1")
-    assertUnsupported("TRUNCATE TABLE my_temp_tab")
+    intercept[NoSuchTableException] {
+      sql("TRUNCATE TABLE my_temp_tab")
+    }
     assertUnsupported("TRUNCATE TABLE my_ext_tab")
     assertUnsupported("TRUNCATE TABLE my_view")
   }
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
index 0b629180d3c5..8cfc89e8b031 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
@@ -85,106 +85,51 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
   test("Issue exceptions for ALTER VIEW on the temporary view") {
     val viewName = "testView"
     withTempView(viewName) {
-      createTempView(viewName)
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER VIEW $viewName SET TBLPROPERTIES ('p' = 'an')")
-      }
-      intercept[NoSuchTableException] {
-        sql(s"ALTER VIEW $viewName UNSET TBLPROPERTIES ('p')")
-      }
+      spark.range(10).createTempView(viewName)
+      assertNoSuchTable(s"ALTER VIEW $viewName SET TBLPROPERTIES ('p' = 'an')")
+      assertNoSuchTable(s"ALTER VIEW $viewName UNSET TBLPROPERTIES ('p')")
     }
   }
 
   test("Issue exceptions for ALTER TABLE on the temporary view") {
     val viewName = "testView"
     withTempView(viewName) {
-      createTempView(viewName)
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $viewName SET SERDE 'whatever'")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $viewName PARTITION (a=1, b=2) SET SERDE 'whatever'")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $viewName SET SERDEPROPERTIES ('p' = 'an')")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $viewName SET LOCATION '/path/to/your/lovely/heart'")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $viewName PARTITION (a='4') SET LOCATION '/path/to/home'")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $viewName ADD IF NOT EXISTS PARTITION (a='4', b='8')")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $viewName DROP PARTITION (a='4', b='8')")
-      }
-
-      intercept[NoSuchTableException] {
-        sql(s"ALTER TABLE $viewName PARTITION (a='4') RENAME TO PARTITION (a='5')")
-      }
-
-      val e = intercept[AnalysisException] {
-        sql(s"ALTER TABLE $viewName RECOVER PARTITIONS")
-      }.getMessage
-      assert(e.contains(
-        "Operation not allowed: ALTER TABLE RECOVER PARTITIONS on temporary tables: `testView`"))
+      spark.range(10).createTempView(viewName)
+      assertNoSuchTable(s"ALTER TABLE $viewName SET SERDE 'whatever'")
+      assertNoSuchTable(s"ALTER TABLE $viewName PARTITION (a=1, b=2) SET SERDE 'whatever'")
+      assertNoSuchTable(s"ALTER TABLE $viewName SET SERDEPROPERTIES ('p' = 'an')")
+      assertNoSuchTable(s"ALTER TABLE $viewName SET LOCATION '/path/to/your/lovely/heart'")
+      assertNoSuchTable(s"ALTER TABLE $viewName PARTITION (a='4') SET LOCATION '/path/to/home'")
+      assertNoSuchTable(s"ALTER TABLE $viewName ADD IF NOT EXISTS PARTITION (a='4', b='8')")
+      assertNoSuchTable(s"ALTER TABLE $viewName DROP PARTITION (a='4', b='8')")
+      assertNoSuchTable(s"ALTER TABLE $viewName PARTITION (a='4') RENAME TO PARTITION (a='5')")
+      assertNoSuchTable(s"ALTER TABLE $viewName RECOVER PARTITIONS")
     }
   }
 
   test("Issue exceptions for other table DDL on the temporary view") {
     val viewName = "testView"
     withTempView(viewName) {
-      createTempView(viewName)
+      spark.range(10).createTempView(viewName)
 
-      var e = intercept[AnalysisException] {
+      val e = intercept[AnalysisException] {
         sql(s"INSERT INTO TABLE $viewName SELECT 1")
       }.getMessage
       assert(e.contains("Inserting into an RDD-based table is not allowed"))
 
       val testData = hiveContext.getHiveFile("data/files/employee.dat").getCanonicalPath
-      e = intercept[AnalysisException] {
-        sql(s"""LOAD DATA LOCAL INPATH "$testData" INTO TABLE $viewName""")
-      }.getMessage
-      assert(e.contains(s"Operation not allowed: LOAD DATA on temporary tables: `$viewName`"))
-
-      e = intercept[AnalysisException] {
-        sql(s"TRUNCATE TABLE $viewName")
-      }.getMessage
-      assert(e.contains(s"Operation not allowed: TRUNCATE TABLE on temporary tables: `$viewName`"))
-
-      e = intercept[AnalysisException] {
-        sql(s"SHOW CREATE TABLE $viewName")
-      }.getMessage
-      assert(e.contains(
-        s"Operation not allowed: SHOW CREATE TABLE on temporary tables: `$viewName`"))
-
-      e = intercept[AnalysisException] {
-        sql(s"SHOW PARTITIONS $viewName")
-      }.getMessage
-      assert(e.contains(s"Operation not allowed: SHOW PARTITIONS on temporary tables: `$viewName`"))
-
-      intercept[NoSuchTableException] {
-        sql(s"ANALYZE TABLE $viewName COMPUTE STATISTICS")
-      }
+      assertNoSuchTable(s"""LOAD DATA LOCAL INPATH "$testData" INTO TABLE $viewName""")
+      assertNoSuchTable(s"TRUNCATE TABLE $viewName")
+      assertNoSuchTable(s"SHOW CREATE TABLE $viewName")
+      assertNoSuchTable(s"SHOW PARTITIONS $viewName")
+      assertNoSuchTable(s"ANALYZE TABLE $viewName COMPUTE STATISTICS")
     }
   }
 
-  private def createTempView(viewName: String): Unit = {
-    sql(
-      s"""
-         |CREATE TEMPORARY VIEW $viewName
-         |AS SELECT 1, 2, 3
-       """.stripMargin)
+  private def assertNoSuchTable(query: String): Unit = {
+    intercept[NoSuchTableException] {
+      sql(query)
+    }
   }
 
   test("error handling: insert/load/truncate table commands against a view") {
@@ -200,12 +145,12 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       e = intercept[AnalysisException] {
         sql(s"""LOAD DATA LOCAL INPATH "$testData" INTO TABLE $viewName""")
       }.getMessage
-      assert(e.contains(s"Target table in LOAD DATA cannot be a view: `$viewName`"))
+      assert(e.contains(s"Target table in LOAD DATA cannot be a view: `default`.`$viewName`"))
 
       e = intercept[AnalysisException] {
         sql(s"TRUNCATE TABLE $viewName")
       }.getMessage
-      assert(e.contains(s"Operation not allowed: TRUNCATE TABLE on views: `$viewName`"))
+      assert(e.contains(s"Operation not allowed: TRUNCATE TABLE on views: `default`.`$viewName`"))
     }
   }
 
@@ -360,13 +305,8 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
   }
 
   test("should not allow ALTER VIEW AS when the view does not exist") {
-    intercept[NoSuchTableException](
-      sql("ALTER VIEW testView AS SELECT 1, 2")
-    )
-
-    intercept[NoSuchTableException](
-      sql("ALTER VIEW default.testView AS SELECT 1, 2")
-    )
+    assertNoSuchTable("ALTER VIEW testView AS SELECT 1, 2")
+    assertNoSuchTable("ALTER VIEW default.testView AS SELECT 1, 2")
   }
 
   test("ALTER VIEW AS should try to alter temp view first if view name has no database part") {

From 7c1a8e53dd1e3b36bf0f20d4ba37a6f2b71cbb2f Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Wed, 14 Sep 2016 16:48:20 -0700
Subject: [PATCH 05/10] fix

---
 .../sql/catalyst/catalog/SessionCatalog.scala | 20 ++++++++--
 .../spark/sql/execution/command/ddl.scala     | 32 +++++++--------
 .../spark/sql/execution/command/tables.scala  | 40 +++++++++----------
 .../sql/hive/execution/HiveCommandSuite.scala | 17 ++++----
 .../sql/hive/execution/HiveDDLSuite.scala     |  2 +-
 .../sql/hive/execution/SQLViewSuite.scala     |  4 +-
 6 files changed, 61 insertions(+), 54 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index 693e17ab5b24..f13e83f9df9c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -246,9 +246,10 @@ class SessionCatalog(
   }
 
   /**
-   * Retrieve the metadata of an existing metastore table.
-   * If no database is specified, assume the table is in the current database.
-   * If the specified table is not found in the database then a [[NoSuchTableException]] is thrown.
+   * Retrieve the metadata of an existing metastore table/view or a temporary view.
+   * If no database is specified, we check whether the corresponding temporary view exists.
+   * If the temporary view does not exist, we assume the table/view is in the current database.
+   * If still not found in the database then a [[NoSuchTableException]] is thrown.
    */
   def getTableMetadata(name: TableIdentifier): CatalogTable = {
     val db = formatDatabaseName(name.database.getOrElse(getCurrentDatabase))
@@ -269,6 +270,19 @@ class SessionCatalog(
     }
   }
 
+  /**
+   * Retrieve the metadata of an existing permanent table/view. If no database is specified,
+   * assume the table/view is in the current database. If the specified table/view is not found
+   * in the database then a [[NoSuchTableException]] is thrown.
+   */
+  def getNonTempTableMetadata(name: TableIdentifier): CatalogTable = {
+    val db = formatDatabaseName(name.database.getOrElse(getCurrentDatabase))
+    val table = formatTableName(name.table)
+    requireDbExists(db)
+    requireTableExists(TableIdentifier(table, Some(db)))
+    externalCatalog.getTable(db, table)
+  }
+
   /**
    * Retrieve the metadata of an existing metastore table.
    * If no database is specified, assume the table is in the current database.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index 8fee2a523389..91ccd30ec042 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -265,15 +265,13 @@ case class AlterTableUnsetPropertiesCommand(
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
     DDLUtils.verifyAlterTableType(catalog, tableName, isView)
-    val db = tableName.database.getOrElse(catalog.getCurrentDatabase)
-    val qualifiedName = TableIdentifier(tableName.table, Some(db))
-    val table = catalog.getTableMetadata(qualifiedName)
+    val table = catalog.getNonTempTableMetadata(tableName)
 
     if (!ifExists) {
       propKeys.foreach { k =>
         if (!table.properties.contains(k)) {
           throw new AnalysisException(
-            s"Attempted to unset non-existent property '$k' in table '$tableName'")
+            s"Attempted to unset non-existent property '$k' in table '${table.identifier}'")
         }
       }
     }
@@ -307,7 +305,7 @@ case class AlterTableSerDePropertiesCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getTableMetadata(tableName)
+    val table = catalog.getNonTempTableMetadata(tableName)
     // For datasource tables, disallow setting serde or specifying partition
     if (partSpec.isDefined && DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException("Operation not allowed: ALTER TABLE SET " +
@@ -325,11 +323,11 @@ case class AlterTableSerDePropertiesCommand(
       catalog.alterTable(newTable)
     } else {
       val spec = partSpec.get
-      val part = catalog.getPartition(tableName, spec)
+      val part = catalog.getPartition(table.identifier, spec)
       val newPart = part.copy(storage = part.storage.copy(
         serde = serdeClassName.orElse(part.storage.serde),
         properties = part.storage.properties ++ serdeProperties.getOrElse(Map())))
-      catalog.alterPartitions(tableName, Seq(newPart))
+      catalog.alterPartitions(table.identifier, Seq(newPart))
     }
     Seq.empty[Row]
   }
@@ -356,7 +354,7 @@ case class AlterTableAddPartitionCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getTableMetadata(tableName)
+    val table = catalog.getNonTempTableMetadata(tableName)
     if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
         "ALTER TABLE ADD PARTITION is not allowed for tables defined using the datasource API")
@@ -365,7 +363,7 @@ case class AlterTableAddPartitionCommand(
       // inherit table storage format (possibly except for location)
       CatalogTablePartition(spec, table.storage.copy(locationUri = location))
     }
-    catalog.createPartitions(tableName, parts, ignoreIfExists = ifNotExists)
+    catalog.createPartitions(table.identifier, parts, ignoreIfExists = ifNotExists)
     Seq.empty[Row]
   }
 
@@ -416,12 +414,12 @@ case class AlterTableDropPartitionCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getTableMetadata(tableName)
+    val table = catalog.getNonTempTableMetadata(tableName)
     if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
         "ALTER TABLE DROP PARTITIONS is not allowed for tables defined using the datasource API")
     }
-    catalog.dropPartitions(tableName, specs, ignoreIfNotExists = ifExists, purge = purge)
+    catalog.dropPartitions(table.identifier, specs, ignoreIfNotExists = ifExists, purge = purge)
     Seq.empty[Row]
   }
 
@@ -470,9 +468,9 @@ case class AlterTableRecoverPartitionsCommand(
 
   override def run(spark: SparkSession): Seq[Row] = {
     val catalog = spark.sessionState.catalog
-    val db = tableName.database.getOrElse(catalog.getCurrentDatabase)
-    val qualifiedName = TableIdentifier(tableName.table, Some(db))
-    val table = catalog.getTableMetadata(qualifiedName)
+    val table = catalog.getNonTempTableMetadata(tableName)
+    val qualifiedName = table.identifier.quotedString
+
     if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
         s"Operation not allowed: $cmd on datasource tables: $qualifiedName")
@@ -647,11 +645,11 @@ case class AlterTableSetLocationCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getTableMetadata(tableName)
+    val table = catalog.getNonTempTableMetadata(tableName)
     partitionSpec match {
       case Some(spec) =>
         // Partition spec is specified, so we set the location only for this partition
-        val part = catalog.getPartition(tableName, spec)
+        val part = catalog.getPartition(table.identifier, spec)
         val newPart =
           if (DDLUtils.isDatasourceTable(table)) {
             throw new AnalysisException(
@@ -660,7 +658,7 @@ case class AlterTableSetLocationCommand(
           } else {
             part.copy(storage = part.storage.copy(locationUri = Some(location)))
           }
-        catalog.alterPartitions(tableName, Seq(newPart))
+        catalog.alterPartitions(table.identifier, Seq(newPart))
       case None =>
         // No partition spec is specified, so we set the location for the table itself
         val newTable =
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 5c599f003a01..d498e4769eca 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -214,9 +214,8 @@ case class LoadDataCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val db = table.database.getOrElse(catalog.getCurrentDatabase)
-    val qualifiedName = TableIdentifier(table.table, Some(db))
-    val targetTable = catalog.getTableMetadata(qualifiedName)
+    val targetTable = catalog.getNonTempTableMetadata(table)
+    val qualifiedName = targetTable.identifier.quotedString
 
     if (targetTable.tableType == CatalogTableType.VIEW) {
       throw new AnalysisException(s"Target table in LOAD DATA cannot be a view: $qualifiedName")
@@ -334,9 +333,8 @@ case class TruncateTableCommand(
 
   override def run(spark: SparkSession): Seq[Row] = {
     val catalog = spark.sessionState.catalog
-    val db = tableName.database.getOrElse(catalog.getCurrentDatabase)
-    val qualifiedName = TableIdentifier(tableName.table, Some(db))
-    val table = catalog.getTableMetadata(qualifiedName)
+    val table = catalog.getNonTempTableMetadata(tableName)
+    val qualifiedName = table.identifier.quotedString
 
     if (table.tableType == CatalogTableType.EXTERNAL) {
       throw new AnalysisException(
@@ -363,7 +361,7 @@ case class TruncateTableCommand(
       } else if (table.partitionColumnNames.isEmpty) {
         Seq(table.storage.locationUri)
       } else {
-        catalog.listPartitions(qualifiedName, partitionSpec).map(_.storage.locationUri)
+        catalog.listPartitions(table.identifier, partitionSpec).map(_.storage.locationUri)
       }
     val hadoopConf = spark.sessionState.newHadoopConf()
     locations.foreach { location =>
@@ -386,7 +384,7 @@ case class TruncateTableCommand(
     spark.sessionState.refreshTable(tableName.unquotedString)
     // Also try to drop the contents of the table from the columnar cache
     try {
-      spark.sharedState.cacheManager.uncacheQuery(spark.table(qualifiedName.quotedString))
+      spark.sharedState.cacheManager.uncacheQuery(spark.table(table.identifier))
     } catch {
       case NonFatal(e) =>
         log.warn(s"Exception when attempting to uncache table $qualifiedName", e)
@@ -622,7 +620,7 @@ case class ShowColumnsCommand(table: TableIdentifier) extends RunnableCommand {
  * }}}
  */
 case class ShowPartitionsCommand(
-    table: TableIdentifier,
+    tableName: TableIdentifier,
     spec: Option[TablePartitionSpec]) extends RunnableCommand {
   override val output: Seq[Attribute] = {
     AttributeReference("partition", StringType, nullable = false)() :: Nil
@@ -636,9 +634,8 @@ case class ShowPartitionsCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val db = table.database.getOrElse(catalog.getCurrentDatabase)
-    val qualifiedName = TableIdentifier(table.table, Some(db))
-    val tab = catalog.getTableMetadata(qualifiedName)
+    val table = catalog.getNonTempTableMetadata(tableName)
+    val qualifiedName = table.identifier.quotedString
 
     /**
      * Validate and throws an [[AnalysisException]] exception under the following conditions:
@@ -646,19 +643,18 @@ case class ShowPartitionsCommand(
      * 2. If it is a datasource table.
      * 3. If it is a view.
      */
-    if (tab.tableType == VIEW) {
-      throw new AnalysisException(
-        s"SHOW PARTITIONS is not allowed on a view: ${tab.qualifiedName}")
+    if (table.tableType == VIEW) {
+      throw new AnalysisException(s"SHOW PARTITIONS is not allowed on a view: $qualifiedName")
     }
 
-    if (tab.partitionColumnNames.isEmpty) {
+    if (table.partitionColumnNames.isEmpty) {
       throw new AnalysisException(
-        s"SHOW PARTITIONS is not allowed on a table that is not partitioned: ${tab.qualifiedName}")
+        s"SHOW PARTITIONS is not allowed on a table that is not partitioned: $qualifiedName")
     }
 
-    if (DDLUtils.isDatasourceTable(tab)) {
+    if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
-        s"SHOW PARTITIONS is not allowed on a datasource table: ${tab.qualifiedName}")
+        s"SHOW PARTITIONS is not allowed on a datasource table: $qualifiedName")
     }
 
     /**
@@ -667,7 +663,7 @@ case class ShowPartitionsCommand(
      * thrown if the partitioning spec is invalid.
      */
     if (spec.isDefined) {
-      val badColumns = spec.get.keySet.filterNot(tab.partitionColumnNames.contains)
+      val badColumns = spec.get.keySet.filterNot(table.partitionColumnNames.contains)
       if (badColumns.nonEmpty) {
         val badCols = badColumns.mkString("[", ", ", "]")
         throw new AnalysisException(
@@ -675,8 +671,8 @@ case class ShowPartitionsCommand(
       }
     }
 
-    val partNames = catalog.listPartitions(table, spec).map { p =>
-      getPartName(p.spec, tab.partitionColumnNames)
+    val partNames = catalog.listPartitions(tableName, spec).map { p =>
+      getPartName(p.spec, table.partitionColumnNames)
     }
 
     partNames.map(Row(_))
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala
index 5db15cff16bc..b2103b3bfc36 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala
@@ -406,25 +406,24 @@ class HiveCommandSuite extends QueryTest with SQLTestUtils with TestHiveSingleto
           |USING org.apache.spark.sql.parquet.DefaultSource
         """.stripMargin)
       // An empty sequence of row is returned for session temporary table.
-      val message1 = intercept[AnalysisException] {
+      intercept[NoSuchTableException] {
         sql("SHOW PARTITIONS parquet_temp")
-      }.getMessage
-      assert(message1.contains("Operation not allowed: SHOW PARTITIONS on temporary tables"))
+      }
 
-      val message2 = intercept[AnalysisException] {
+      val message1 = intercept[AnalysisException] {
         sql("SHOW PARTITIONS parquet_tab3")
       }.getMessage
-      assert(message2.contains("not allowed on a table that is not partitioned"))
+      assert(message1.contains("not allowed on a table that is not partitioned"))
 
-      val message3 = intercept[AnalysisException] {
+      val message2 = intercept[AnalysisException] {
         sql("SHOW PARTITIONS parquet_tab4 PARTITION(abcd=2015, xyz=1)")
       }.getMessage
-      assert(message3.contains("Non-partitioning column(s) [abcd, xyz] are specified"))
+      assert(message2.contains("Non-partitioning column(s) [abcd, xyz] are specified"))
 
-      val message4 = intercept[AnalysisException] {
+      val message3 = intercept[AnalysisException] {
         sql("SHOW PARTITIONS parquet_view1")
       }.getMessage
-      assert(message4.contains("is not allowed on a view"))
+      assert(message3.contains("is not allowed on a view"))
     }
   }
 
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index 3cba5b2a097f..099de2a1ccbe 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -300,7 +300,7 @@ class HiveDDLSuite
           sql(s"ALTER VIEW $viewName UNSET TBLPROPERTIES ('p')")
         }.getMessage
         assert(message.contains(
-          "Attempted to unset non-existent property 'p' in table '`view1`'"))
+          "Attempted to unset non-existent property 'p' in table '`default`.`view1`'"))
       }
     }
   }
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
index 8cfc89e8b031..a215c70da0c5 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLViewSuite.scala
@@ -145,12 +145,12 @@ class SQLViewSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       e = intercept[AnalysisException] {
         sql(s"""LOAD DATA LOCAL INPATH "$testData" INTO TABLE $viewName""")
       }.getMessage
-      assert(e.contains(s"Target table in LOAD DATA cannot be a view: `default`.`$viewName`"))
+      assert(e.contains(s"Target table in LOAD DATA cannot be a view: `default`.`testview`"))
 
       e = intercept[AnalysisException] {
         sql(s"TRUNCATE TABLE $viewName")
       }.getMessage
-      assert(e.contains(s"Operation not allowed: TRUNCATE TABLE on views: `default`.`$viewName`"))
+      assert(e.contains(s"Operation not allowed: TRUNCATE TABLE on views: `default`.`testview`"))
     }
   }
 

From a01f6b39f2073390ebf2445a1eca1c3925d24841 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Wed, 14 Sep 2016 23:44:24 -0700
Subject: [PATCH 06/10] address comments

---
 .../analysis/NoSuchItemException.scala        |  3 ++
 .../sql/catalyst/catalog/SessionCatalog.scala | 33 +++++++------------
 .../catalog/SessionCatalogSuite.scala         | 13 ++++++--
 .../spark/sql/execution/command/ddl.scala     | 12 +++----
 .../spark/sql/execution/command/tables.scala  | 32 ++++++++++++------
 .../spark/sql/internal/CatalogImpl.scala      |  7 +++-
 .../sql/hive/execution/HiveDDLSuite.scala     |  3 +-
 7 files changed, 60 insertions(+), 43 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala
index 8febdcaee829..9f2b94eb1d5b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala
@@ -30,6 +30,9 @@ class NoSuchDatabaseException(db: String) extends AnalysisException(s"Database '
 class NoSuchTableException(db: String, table: String)
   extends AnalysisException(s"Table or view '$table' not found in database '$db'")
 
+class NoSuchTempViewException(table: String)
+  extends AnalysisException(s"Temporary view '$table' not found")
+
 class NoSuchPartitionException(
     db: String,
     table: String,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index f13e83f9df9c..55493449467c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -246,28 +246,19 @@ class SessionCatalog(
   }
 
   /**
-   * Retrieve the metadata of an existing metastore table/view or a temporary view.
-   * If no database is specified, we check whether the corresponding temporary view exists.
-   * If the temporary view does not exist, we assume the table/view is in the current database.
-   * If still not found in the database then a [[NoSuchTableException]] is thrown.
+   * Retrieve the metadata of an existing temporary view.
+   * If the temporary view does not exist, a [[NoSuchTempViewException]] is thrown.
    */
-  def getTableMetadata(name: TableIdentifier): CatalogTable = {
-    val db = formatDatabaseName(name.database.getOrElse(getCurrentDatabase))
-    val table = formatTableName(name.table)
-    val tid = TableIdentifier(table)
-    if (isTemporaryTable(name)) {
-      CatalogTable(
-        identifier = tid,
-        tableType = CatalogTableType.VIEW,
-        storage = CatalogStorageFormat.empty,
-        schema = tempTables(table).output.toStructType,
-        properties = Map(),
-        viewText = None)
-    } else {
-      requireDbExists(db)
-      requireTableExists(TableIdentifier(table, Some(db)))
-      externalCatalog.getTable(db, table)
+  def getTempViewMetadata(name: String): CatalogTable = {
+    val table = formatTableName(name)
+    if (!tempTables.contains(table)) {
+      throw new NoSuchTempViewException(table)
     }
+    CatalogTable(
+      identifier = TableIdentifier(table),
+      tableType = CatalogTableType.VIEW,
+      storage = CatalogStorageFormat.empty,
+      schema = tempTables(table).output.toStructType)
   }
 
   /**
@@ -275,7 +266,7 @@ class SessionCatalog(
    * assume the table/view is in the current database. If the specified table/view is not found
    * in the database then a [[NoSuchTableException]] is thrown.
    */
-  def getNonTempTableMetadata(name: TableIdentifier): CatalogTable = {
+  def getTableMetadata(name: TableIdentifier): CatalogTable = {
     val db = formatDatabaseName(name.database.getOrElse(getCurrentDatabase))
     val table = formatTableName(name.table)
     requireDbExists(db)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
index 012df629bbde..a745333201fe 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
@@ -444,7 +444,7 @@ class SessionCatalogSuite extends SparkFunSuite {
     assert(!catalog.tableExists(TableIdentifier("view1", Some("default"))))
   }
 
-  test("getTableMetadata on temporary views") {
+  test("getTableMetadata and getTempViewMetadata on temporary views") {
     val catalog = new SessionCatalog(newBasicCatalog())
     val tempTable = Range(1, 10, 2, 10)
     val m = intercept[AnalysisException] {
@@ -457,9 +457,16 @@ class SessionCatalogSuite extends SparkFunSuite {
     }.getMessage
     assert(m2.contains("Table or view 'view1' not found in database 'default'"))
 
+    intercept[NoSuchTempViewException] {
+      catalog.getTempViewMetadata("view1")
+    }.getMessage
+
     catalog.createTempView("view1", tempTable, overrideIfExists = false)
-    assert(catalog.getTableMetadata(TableIdentifier("view1")).identifier.table == "view1")
-    assert(catalog.getTableMetadata(TableIdentifier("view1")).schema(0).name == "id")
+    assert(catalog.getTempViewMetadata("view1").identifier === TableIdentifier("view1"))
+
+    intercept[NoSuchTableException] {
+      catalog.getTableMetadata(TableIdentifier("view1"))
+    }.getMessage
 
     val m3 = intercept[AnalysisException] {
       catalog.getTableMetadata(TableIdentifier("view1", Some("default")))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index 91ccd30ec042..565da30944cf 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -265,7 +265,7 @@ case class AlterTableUnsetPropertiesCommand(
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
     DDLUtils.verifyAlterTableType(catalog, tableName, isView)
-    val table = catalog.getNonTempTableMetadata(tableName)
+    val table = catalog.getTableMetadata(tableName)
 
     if (!ifExists) {
       propKeys.foreach { k =>
@@ -305,7 +305,7 @@ case class AlterTableSerDePropertiesCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getNonTempTableMetadata(tableName)
+    val table = catalog.getTableMetadata(tableName)
     // For datasource tables, disallow setting serde or specifying partition
     if (partSpec.isDefined && DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException("Operation not allowed: ALTER TABLE SET " +
@@ -354,7 +354,7 @@ case class AlterTableAddPartitionCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getNonTempTableMetadata(tableName)
+    val table = catalog.getTableMetadata(tableName)
     if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
         "ALTER TABLE ADD PARTITION is not allowed for tables defined using the datasource API")
@@ -414,7 +414,7 @@ case class AlterTableDropPartitionCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getNonTempTableMetadata(tableName)
+    val table = catalog.getTableMetadata(tableName)
     if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
         "ALTER TABLE DROP PARTITIONS is not allowed for tables defined using the datasource API")
@@ -468,7 +468,7 @@ case class AlterTableRecoverPartitionsCommand(
 
   override def run(spark: SparkSession): Seq[Row] = {
     val catalog = spark.sessionState.catalog
-    val table = catalog.getNonTempTableMetadata(tableName)
+    val table = catalog.getTableMetadata(tableName)
     val qualifiedName = table.identifier.quotedString
 
     if (DDLUtils.isDatasourceTable(table)) {
@@ -645,7 +645,7 @@ case class AlterTableSetLocationCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getNonTempTableMetadata(tableName)
+    val table = catalog.getTableMetadata(tableName)
     partitionSpec match {
       case Some(spec) =>
         // Partition spec is specified, so we set the location only for this partition
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index d498e4769eca..0c676c5e8122 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -64,7 +64,11 @@ case class CreateTableLikeCommand(
         s"Source table in CREATE TABLE LIKE does not exist: '$sourceTable'")
     }
 
-    val sourceTableDesc = catalog.getTableMetadata(sourceTable)
+    val sourceTableDesc = if (catalog.isTemporaryTable(sourceTable)) {
+      catalog.getTempViewMetadata(sourceTable.table)
+    } else {
+      catalog.getTableMetadata(sourceTable)
+    }
 
     // Storage format
     val newStorage =
@@ -176,7 +180,11 @@ case class AlterTableRenameCommand(
         }
       }
       // For datasource tables, we also need to update the "path" serde property
-      val table = catalog.getTableMetadata(oldName)
+      val table = if (catalog.isTemporaryTable(oldName)) {
+        catalog.getTempViewMetadata(oldName.table)
+      } else {
+        catalog.getTableMetadata(oldName)
+      }
       if (DDLUtils.isDatasourceTable(table) && table.tableType == CatalogTableType.MANAGED) {
         val newPath = catalog.defaultTablePath(newTblName)
         val newTable = table.withNewStorage(
@@ -214,7 +222,7 @@ case class LoadDataCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val targetTable = catalog.getNonTempTableMetadata(table)
+    val targetTable = catalog.getTableMetadata(table)
     val qualifiedName = targetTable.identifier.quotedString
 
     if (targetTable.tableType == CatalogTableType.VIEW) {
@@ -333,7 +341,7 @@ case class TruncateTableCommand(
 
   override def run(spark: SparkSession): Seq[Row] = {
     val catalog = spark.sessionState.catalog
-    val table = catalog.getNonTempTableMetadata(tableName)
+    val table = catalog.getTableMetadata(tableName)
     val qualifiedName = table.identifier.quotedString
 
     if (table.tableType == CatalogTableType.EXTERNAL) {
@@ -592,13 +600,19 @@ case class ShowTablePropertiesCommand(table: TableIdentifier, propertyKey: Optio
  *   SHOW COLUMNS (FROM | IN) table_identifier [(FROM | IN) database];
  * }}}
  */
-case class ShowColumnsCommand(table: TableIdentifier) extends RunnableCommand {
+case class ShowColumnsCommand(tableName: TableIdentifier) extends RunnableCommand {
   override val output: Seq[Attribute] = {
     AttributeReference("col_name", StringType, nullable = false)() :: Nil
   }
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
-    sparkSession.sessionState.catalog.getTableMetadata(table).schema.map { c =>
+    val catalog = sparkSession.sessionState.catalog
+    val table = if (catalog.isTemporaryTable(tableName)) {
+      catalog.getTempViewMetadata(tableName.table)
+    } else {
+      catalog.getTableMetadata(tableName)
+    }
+    table.schema.map { c =>
       Row(c.name)
     }
   }
@@ -634,7 +648,7 @@ case class ShowPartitionsCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = catalog.getNonTempTableMetadata(tableName)
+    val table = catalog.getTableMetadata(tableName)
     val qualifiedName = table.identifier.quotedString
 
     /**
@@ -686,9 +700,7 @@ case class ShowCreateTableCommand(table: TableIdentifier) extends RunnableComman
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val db = table.database.getOrElse(catalog.getCurrentDatabase)
-    val qualifiedName = TableIdentifier(table.table, Some(db))
-    val tableMetadata = catalog.getTableMetadata(qualifiedName)
+    val tableMetadata = catalog.getTableMetadata(table)
 
     // TODO: unify this after we unify the CREATE TABLE syntax for hive serde and data source table.
     val stmt = if (DDLUtils.isDatasourceTable(tableMetadata)) {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
index 1f87f0e73a3b..5be850329ccf 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
@@ -151,7 +151,12 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
   }
 
   private def listColumns(tableIdentifier: TableIdentifier): Dataset[Column] = {
-    val tableMetadata = sessionCatalog.getTableMetadata(tableIdentifier)
+    val tableMetadata = if (sessionCatalog.isTemporaryTable(tableIdentifier)) {
+      sessionCatalog.getTempViewMetadata(tableIdentifier.table)
+    } else {
+      sessionCatalog.getTableMetadata(tableIdentifier)
+    }
+
     val partitionColumnNames = tableMetadata.partitionColumnNames.toSet
     val bucketColumnNames = tableMetadata.bucketSpec.map(_.bucketColumnNames).getOrElse(Nil).toSet
     val columns = tableMetadata.schema.map { c =>
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index 099de2a1ccbe..b179c9b72fdc 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -671,8 +671,7 @@ class HiveDDLSuite
           .createTempView(sourceViewName)
         sql(s"CREATE TABLE $targetTabName LIKE $sourceViewName")
 
-        val sourceTable = spark.sessionState.catalog.getTableMetadata(
-          TableIdentifier(sourceViewName, None))
+        val sourceTable = spark.sessionState.catalog.getTempViewMetadata(sourceViewName)
         val targetTable = spark.sessionState.catalog.getTableMetadata(
           TableIdentifier(targetTabName, Some("default")))
 

From f305c4c2ae1bfb2c4d94d1e17942ff6273c3da2b Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Sun, 18 Sep 2016 20:09:02 -0700
Subject: [PATCH 07/10] address comments

---
 .../sql/catalyst/catalog/SessionCatalog.scala | 21 ++++++++++++-------
 .../spark/sql/execution/command/tables.scala  | 14 +++++++------
 .../spark/sql/internal/CatalogImpl.scala      |  7 ++++---
 3 files changed, 26 insertions(+), 16 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index 52a6b307b9c3..eeec7fc8b92b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -250,15 +250,22 @@ class SessionCatalog(
    * If the temporary view does not exist, a [[NoSuchTempViewException]] is thrown.
    */
   def getTempViewMetadata(name: String): CatalogTable = {
+    getTempViewMetadataOption(name).getOrElse(throw new NoSuchTempViewException(name))
+  }
+
+  /**
+   * Retrieve the metadata of an existing temporary view.
+   * If the temporary view does not exist, return None.
+   */
+  def getTempViewMetadataOption(name: String): Option[CatalogTable] = synchronized {
     val table = formatTableName(name)
-    if (!tempTables.contains(table)) {
-      throw new NoSuchTempViewException(table)
+    getTempView(table).map { plan =>
+      CatalogTable(
+        identifier = TableIdentifier(table),
+        tableType = CatalogTableType.VIEW,
+        storage = CatalogStorageFormat.empty,
+        schema = plan.output.toStructType)
     }
-    CatalogTable(
-      identifier = TableIdentifier(table),
-      tableType = CatalogTableType.VIEW,
-      storage = CatalogStorageFormat.empty,
-      schema = tempTables(table).output.toStructType)
   }
 
   /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 42fdd68a495f..76f165a65f34 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -64,10 +64,11 @@ case class CreateTableLikeCommand(
         s"Source table in CREATE TABLE LIKE does not exist: '$sourceTable'")
     }
 
-    val sourceTableDesc = if (catalog.isTemporaryTable(sourceTable)) {
-      catalog.getTempViewMetadata(sourceTable.table)
-    } else {
+    val sourceTableDesc = if (sourceTable.database.isDefined) {
       catalog.getTableMetadata(sourceTable)
+    } else {
+      catalog.getTempViewMetadataOption(sourceTable.table)
+        .getOrElse(catalog.getTableMetadata(sourceTable))
     }
 
     // Storage format
@@ -602,10 +603,11 @@ case class ShowColumnsCommand(tableName: TableIdentifier) extends RunnableComman
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val table = if (catalog.isTemporaryTable(tableName)) {
-      catalog.getTempViewMetadata(tableName.table)
-    } else {
+    val table = if (tableName.database.isDefined) {
       catalog.getTableMetadata(tableName)
+    } else {
+      catalog.getTempViewMetadataOption(tableName.table)
+        .getOrElse(catalog.getTableMetadata(tableName))
     }
     table.schema.map { c =>
       Row(c.name)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
index 01e8a7c31b8e..cc61f37e4099 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
@@ -151,10 +151,11 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
   }
 
   private def listColumns(tableIdentifier: TableIdentifier): Dataset[Column] = {
-    val tableMetadata = if (sessionCatalog.isTemporaryTable(tableIdentifier)) {
-      sessionCatalog.getTempViewMetadata(tableIdentifier.table)
-    } else {
+    val tableMetadata = if (tableIdentifier.database.isDefined) {
       sessionCatalog.getTableMetadata(tableIdentifier)
+    } else {
+      sessionCatalog.getTempViewMetadataOption(tableIdentifier.table)
+        .getOrElse(sessionCatalog.getTableMetadata(tableIdentifier))
     }
 
     val partitionColumnNames = tableMetadata.partitionColumnNames.toSet

From c662d2c0e69e48af45a0fc921c8d29e8971810a2 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Mon, 19 Sep 2016 13:12:10 -0700
Subject: [PATCH 08/10] address comments.

---
 .../analysis/NoSuchItemException.scala        |  3 --
 .../sql/catalyst/catalog/SessionCatalog.scala | 38 ++++++++-----------
 .../catalog/SessionCatalogSuite.scala         | 10 ++---
 .../command/AnalyzeTableCommand.scala         |  4 +-
 .../spark/sql/execution/command/ddl.scala     | 10 ++---
 .../spark/sql/execution/command/tables.scala  | 38 +++++++++----------
 .../sql/hive/execution/HiveDDLSuite.scala     |  2 +-
 7 files changed, 47 insertions(+), 58 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala
index 9f2b94eb1d5b..8febdcaee829 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NoSuchItemException.scala
@@ -30,9 +30,6 @@ class NoSuchDatabaseException(db: String) extends AnalysisException(s"Database '
 class NoSuchTableException(db: String, table: String)
   extends AnalysisException(s"Table or view '$table' not found in database '$db'")
 
-class NoSuchTempViewException(table: String)
-  extends AnalysisException(s"Temporary view '$table' not found")
-
 class NoSuchPartitionException(
     db: String,
     table: String,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index eeec7fc8b92b..cce901854d1d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -245,29 +245,6 @@ class SessionCatalog(
     externalCatalog.alterTable(newTableDefinition)
   }
 
-  /**
-   * Retrieve the metadata of an existing temporary view.
-   * If the temporary view does not exist, a [[NoSuchTempViewException]] is thrown.
-   */
-  def getTempViewMetadata(name: String): CatalogTable = {
-    getTempViewMetadataOption(name).getOrElse(throw new NoSuchTempViewException(name))
-  }
-
-  /**
-   * Retrieve the metadata of an existing temporary view.
-   * If the temporary view does not exist, return None.
-   */
-  def getTempViewMetadataOption(name: String): Option[CatalogTable] = synchronized {
-    val table = formatTableName(name)
-    getTempView(table).map { plan =>
-      CatalogTable(
-        identifier = TableIdentifier(table),
-        tableType = CatalogTableType.VIEW,
-        storage = CatalogStorageFormat.empty,
-        schema = plan.output.toStructType)
-    }
-  }
-
   /**
    * Retrieve the metadata of an existing permanent table/view. If no database is specified,
    * assume the table/view is in the current database. If the specified table/view is not found
@@ -369,6 +346,21 @@ class SessionCatalog(
     tempTables.remove(formatTableName(name))
   }
 
+  /**
+   * Retrieve the metadata of an existing temporary view.
+   * If the temporary view does not exist, return None.
+   */
+  def getTempViewMetadataOption(name: String): Option[CatalogTable] = synchronized {
+    val table = formatTableName(name)
+    getTempView(table).map { plan =>
+      CatalogTable(
+        identifier = TableIdentifier(table),
+        tableType = CatalogTableType.VIEW,
+        storage = CatalogStorageFormat.empty,
+        schema = plan.output.toStructType)
+    }
+  }
+
   // -------------------------------------------------------------
   // | Methods that interact with temporary and metastore tables |
   // -------------------------------------------------------------
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
index 342969ca33de..86ac9c01fd74 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
@@ -444,7 +444,7 @@ class SessionCatalogSuite extends SparkFunSuite {
     assert(!catalog.tableExists(TableIdentifier("view1", Some("default"))))
   }
 
-  test("getTableMetadata and getTempViewMetadata on temporary views") {
+  test("getTableMetadata and getTempViewMetadataOption on temporary views") {
     val catalog = new SessionCatalog(newBasicCatalog())
     val tempTable = Range(1, 10, 2, 10)
     val m = intercept[AnalysisException] {
@@ -457,12 +457,12 @@ class SessionCatalogSuite extends SparkFunSuite {
     }.getMessage
     assert(m2.contains("Table or view 'view1' not found in database 'default'"))
 
-    intercept[NoSuchTempViewException] {
-      catalog.getTempViewMetadata("view1")
-    }.getMessage
+    assert(catalog.getTempViewMetadataOption("view1").isEmpty,
+      "the temporary view `view1` should not exist")
 
     catalog.createTempView("view1", tempTable, overrideIfExists = false)
-    assert(catalog.getTempViewMetadata("view1").identifier === TableIdentifier("view1"))
+    assert(catalog.getTempViewMetadataOption("view1").get.identifier === TableIdentifier("view1"),
+      "the temporary view `view1` should exist")
 
     intercept[NoSuchTableException] {
       catalog.getTableMetadata(TableIdentifier("view1"))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala
index ad27e0a53f35..40aecafecf5b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala
@@ -39,8 +39,8 @@ case class AnalyzeTableCommand(tableName: String, noscan: Boolean = true) extend
     val sessionState = sparkSession.sessionState
     val tableIdent = sessionState.sqlParser.parseTableIdentifier(tableName)
     val db = tableIdent.database.getOrElse(sessionState.catalog.getCurrentDatabase)
-    val qualifiedName = TableIdentifier(tableIdent.table, Some(db))
-    val relation = EliminateSubqueryAliases(sessionState.catalog.lookupRelation(qualifiedName))
+    val tableIdentwithDB = TableIdentifier(tableIdent.table, Some(db))
+    val relation = EliminateSubqueryAliases(sessionState.catalog.lookupRelation(tableIdentwithDB))
 
     relation match {
       case relation: CatalogRelation =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
index 0d5fdf762b6b..b57b2d280d8f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
@@ -472,19 +472,19 @@ case class AlterTableRecoverPartitionsCommand(
   override def run(spark: SparkSession): Seq[Row] = {
     val catalog = spark.sessionState.catalog
     val table = catalog.getTableMetadata(tableName)
-    val qualifiedName = table.identifier.quotedString
+    val tableIdentWithDB = table.identifier.quotedString
     DDLUtils.verifyAlterTableType(catalog, table, isView = false)
     if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
-        s"Operation not allowed: $cmd on datasource tables: $qualifiedName")
+        s"Operation not allowed: $cmd on datasource tables: $tableIdentWithDB")
     }
     if (table.partitionColumnNames.isEmpty) {
       throw new AnalysisException(
-        s"Operation not allowed: $cmd only works on partitioned tables: $qualifiedName")
+        s"Operation not allowed: $cmd only works on partitioned tables: $tableIdentWithDB")
     }
     if (table.storage.locationUri.isEmpty) {
-      throw new AnalysisException(
-        s"Operation not allowed: $cmd only works on table with location provided: $qualifiedName")
+      throw new AnalysisException(s"Operation not allowed: $cmd only works on table with " +
+        s"location provided: $tableIdentWithDB")
     }
 
     val root = new Path(table.storage.locationUri.get)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 76f165a65f34..90a5ab7db7df 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -219,37 +219,37 @@ case class LoadDataCommand(
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
     val targetTable = catalog.getTableMetadata(table)
-    val qualifiedName = targetTable.identifier.quotedString
+    val tableIdentwithDB = targetTable.identifier.quotedString
 
     if (targetTable.tableType == CatalogTableType.VIEW) {
-      throw new AnalysisException(s"Target table in LOAD DATA cannot be a view: $qualifiedName")
+      throw new AnalysisException(s"Target table in LOAD DATA cannot be a view: $tableIdentwithDB")
     }
     if (DDLUtils.isDatasourceTable(targetTable)) {
       throw new AnalysisException(
-        s"LOAD DATA is not supported for datasource tables: $qualifiedName")
+        s"LOAD DATA is not supported for datasource tables: $tableIdentwithDB")
     }
     if (targetTable.partitionColumnNames.nonEmpty) {
       if (partition.isEmpty) {
-        throw new AnalysisException(s"LOAD DATA target table $qualifiedName is partitioned, " +
+        throw new AnalysisException(s"LOAD DATA target table $tableIdentwithDB is partitioned, " +
           s"but no partition spec is provided")
       }
       if (targetTable.partitionColumnNames.size != partition.get.size) {
-        throw new AnalysisException(s"LOAD DATA target table $qualifiedName is partitioned, " +
+        throw new AnalysisException(s"LOAD DATA target table $tableIdentwithDB is partitioned, " +
           s"but number of columns in provided partition spec (${partition.get.size}) " +
           s"do not match number of partitioned columns in table " +
           s"(s${targetTable.partitionColumnNames.size})")
       }
       partition.get.keys.foreach { colName =>
         if (!targetTable.partitionColumnNames.contains(colName)) {
-          throw new AnalysisException(s"LOAD DATA target table $qualifiedName is partitioned, " +
+          throw new AnalysisException(s"LOAD DATA target table $tableIdentwithDB is partitioned, " +
             s"but the specified partition spec refers to a column that is not partitioned: " +
             s"'$colName'")
         }
       }
     } else {
       if (partition.nonEmpty) {
-        throw new AnalysisException(s"LOAD DATA target table $qualifiedName is not partitioned, " +
-          s"but a partition spec was provided.")
+        throw new AnalysisException(s"LOAD DATA target table $tableIdentwithDB is not " +
+          s"partitioned, but a partition spec was provided.")
       }
     }
 
@@ -338,26 +338,26 @@ case class TruncateTableCommand(
   override def run(spark: SparkSession): Seq[Row] = {
     val catalog = spark.sessionState.catalog
     val table = catalog.getTableMetadata(tableName)
-    val qualifiedName = table.identifier.quotedString
+    val tableIdentwithDB = table.identifier.quotedString
 
     if (table.tableType == CatalogTableType.EXTERNAL) {
       throw new AnalysisException(
-        s"Operation not allowed: TRUNCATE TABLE on external tables: $qualifiedName")
+        s"Operation not allowed: TRUNCATE TABLE on external tables: $tableIdentwithDB")
     }
     if (table.tableType == CatalogTableType.VIEW) {
       throw new AnalysisException(
-        s"Operation not allowed: TRUNCATE TABLE on views: $qualifiedName")
+        s"Operation not allowed: TRUNCATE TABLE on views: $tableIdentwithDB")
     }
     val isDatasourceTable = DDLUtils.isDatasourceTable(table)
     if (isDatasourceTable && partitionSpec.isDefined) {
       throw new AnalysisException(
         s"Operation not allowed: TRUNCATE TABLE ... PARTITION is not supported " +
-        s"for tables created using the data sources API: $qualifiedName")
+        s"for tables created using the data sources API: $tableIdentwithDB")
     }
     if (table.partitionColumnNames.isEmpty && partitionSpec.isDefined) {
       throw new AnalysisException(
         s"Operation not allowed: TRUNCATE TABLE ... PARTITION is not supported " +
-        s"for tables that are not partitioned: $qualifiedName")
+        s"for tables that are not partitioned: $tableIdentwithDB")
     }
     val locations =
       if (isDatasourceTable) {
@@ -378,7 +378,7 @@ case class TruncateTableCommand(
         } catch {
           case NonFatal(e) =>
             throw new AnalysisException(
-              s"Failed to truncate table $qualifiedName when removing data of the path: $path " +
+              s"Failed to truncate table $tableIdentwithDB when removing data of the path: $path " +
                 s"because of ${e.toString}")
         }
       }
@@ -391,7 +391,7 @@ case class TruncateTableCommand(
       spark.sharedState.cacheManager.uncacheQuery(spark.table(table.identifier))
     } catch {
       case NonFatal(e) =>
-        log.warn(s"Exception when attempting to uncache table $qualifiedName", e)
+        log.warn(s"Exception when attempting to uncache table $tableIdentwithDB", e)
     }
     Seq.empty[Row]
   }
@@ -646,7 +646,7 @@ case class ShowPartitionsCommand(
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
     val table = catalog.getTableMetadata(tableName)
-    val qualifiedName = table.identifier.quotedString
+    val tableIdentWithDB = table.identifier.quotedString
 
     /**
      * Validate and throws an [[AnalysisException]] exception under the following conditions:
@@ -655,17 +655,17 @@ case class ShowPartitionsCommand(
      * 3. If it is a view.
      */
     if (table.tableType == VIEW) {
-      throw new AnalysisException(s"SHOW PARTITIONS is not allowed on a view: $qualifiedName")
+      throw new AnalysisException(s"SHOW PARTITIONS is not allowed on a view: $tableIdentWithDB")
     }
 
     if (table.partitionColumnNames.isEmpty) {
       throw new AnalysisException(
-        s"SHOW PARTITIONS is not allowed on a table that is not partitioned: $qualifiedName")
+        s"SHOW PARTITIONS is not allowed on a table that is not partitioned: $tableIdentWithDB")
     }
 
     if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
-        s"SHOW PARTITIONS is not allowed on a datasource table: $qualifiedName")
+        s"SHOW PARTITIONS is not allowed on a datasource table: $tableIdentWithDB")
     }
 
     /**
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index 703ad11c258b..69eaef04354e 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -678,7 +678,7 @@ class HiveDDLSuite
           .createTempView(sourceViewName)
         sql(s"CREATE TABLE $targetTabName LIKE $sourceViewName")
 
-        val sourceTable = spark.sessionState.catalog.getTempViewMetadata(sourceViewName)
+        val sourceTable = spark.sessionState.catalog.getTempViewMetadataOption(sourceViewName).get
         val targetTable = spark.sessionState.catalog.getTableMetadata(
           TableIdentifier(targetTabName, Some("default")))
 

From 48ce44e20a3db290c4c563b4d45ec5bfb6a86195 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Mon, 19 Sep 2016 22:48:56 -0700
Subject: [PATCH 09/10] address comments

---
 .../sql/catalyst/catalog/SessionCatalog.scala | 33 ++++++++++---------
 .../catalog/SessionCatalogSuite.scala         | 11 ++++---
 .../spark/sql/execution/command/tables.scala  |  6 ++--
 .../spark/sql/internal/CatalogImpl.scala      |  3 +-
 .../sql/hive/execution/HiveDDLSuite.scala     |  3 +-
 5 files changed, 29 insertions(+), 27 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index cce901854d1d..8c6aa6a913f1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -270,6 +270,24 @@ class SessionCatalog(
     externalCatalog.getTableOption(db, table)
   }
 
+  /**
+   * Retrieve the metadata of an existing temporary view or permanent table/view.
+   * If the temporary view does not exist, tries to get the metadata an existing permanent
+   * table/view. If no database is specified, assume the table/view is in the current database.
+   * If the specified table/view is not found in the database then a [[NoSuchTableException]] is
+   * thrown.
+   */
+  def getTempViewOrPermanentTableMetadata(name: TableIdentifier): CatalogTable = synchronized {
+    val table = formatTableName(name.table)
+    getTempView(table).map { plan =>
+      CatalogTable(
+        identifier = TableIdentifier(table),
+        tableType = CatalogTableType.VIEW,
+        storage = CatalogStorageFormat.empty,
+        schema = plan.output.toStructType)
+    }.getOrElse(getTableMetadata(name))
+  }
+
   /**
    * Load files stored in given path into an existing metastore table.
    * If no database is specified, assume the table is in the current database.
@@ -346,21 +364,6 @@ class SessionCatalog(
     tempTables.remove(formatTableName(name))
   }
 
-  /**
-   * Retrieve the metadata of an existing temporary view.
-   * If the temporary view does not exist, return None.
-   */
-  def getTempViewMetadataOption(name: String): Option[CatalogTable] = synchronized {
-    val table = formatTableName(name)
-    getTempView(table).map { plan =>
-      CatalogTable(
-        identifier = TableIdentifier(table),
-        tableType = CatalogTableType.VIEW,
-        storage = CatalogStorageFormat.empty,
-        schema = plan.output.toStructType)
-    }
-  }
-
   // -------------------------------------------------------------
   // | Methods that interact with temporary and metastore tables |
   // -------------------------------------------------------------
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
index 86ac9c01fd74..e071ca8e1225 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
@@ -444,7 +444,7 @@ class SessionCatalogSuite extends SparkFunSuite {
     assert(!catalog.tableExists(TableIdentifier("view1", Some("default"))))
   }
 
-  test("getTableMetadata and getTempViewMetadataOption on temporary views") {
+  test("getTableMetadata and getTempViewOrPermanentTableMetadata on temporary views") {
     val catalog = new SessionCatalog(newBasicCatalog())
     val tempTable = Range(1, 10, 2, 10)
     val m = intercept[AnalysisException] {
@@ -457,12 +457,13 @@ class SessionCatalogSuite extends SparkFunSuite {
     }.getMessage
     assert(m2.contains("Table or view 'view1' not found in database 'default'"))
 
-    assert(catalog.getTempViewMetadataOption("view1").isEmpty,
-      "the temporary view `view1` should not exist")
+    intercept[NoSuchTableException] {
+      catalog.getTempViewOrPermanentTableMetadata(TableIdentifier("view1"))
+    }.getMessage
 
     catalog.createTempView("view1", tempTable, overrideIfExists = false)
-    assert(catalog.getTempViewMetadataOption("view1").get.identifier === TableIdentifier("view1"),
-      "the temporary view `view1` should exist")
+    assert(catalog.getTempViewOrPermanentTableMetadata(TableIdentifier("view1")).identifier ==
+      TableIdentifier("view1"), "the temporary view `view1` should exist")
 
     intercept[NoSuchTableException] {
       catalog.getTableMetadata(TableIdentifier("view1"))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 90a5ab7db7df..0bafdaa184a0 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -67,8 +67,7 @@ case class CreateTableLikeCommand(
     val sourceTableDesc = if (sourceTable.database.isDefined) {
       catalog.getTableMetadata(sourceTable)
     } else {
-      catalog.getTempViewMetadataOption(sourceTable.table)
-        .getOrElse(catalog.getTableMetadata(sourceTable))
+      catalog.getTempViewOrPermanentTableMetadata(sourceTable)
     }
 
     // Storage format
@@ -606,8 +605,7 @@ case class ShowColumnsCommand(tableName: TableIdentifier) extends RunnableComman
     val table = if (tableName.database.isDefined) {
       catalog.getTableMetadata(tableName)
     } else {
-      catalog.getTempViewMetadataOption(tableName.table)
-        .getOrElse(catalog.getTableMetadata(tableName))
+      catalog.getTempViewOrPermanentTableMetadata(tableName)
     }
     table.schema.map { c =>
       Row(c.name)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
index cc61f37e4099..ee757e7dc166 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
@@ -154,8 +154,7 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
     val tableMetadata = if (tableIdentifier.database.isDefined) {
       sessionCatalog.getTableMetadata(tableIdentifier)
     } else {
-      sessionCatalog.getTempViewMetadataOption(tableIdentifier.table)
-        .getOrElse(sessionCatalog.getTableMetadata(tableIdentifier))
+      sessionCatalog.getTempViewOrPermanentTableMetadata(tableIdentifier)
     }
 
     val partitionColumnNames = tableMetadata.partitionColumnNames.toSet
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index 69eaef04354e..c927e5d802c9 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -678,7 +678,8 @@ class HiveDDLSuite
           .createTempView(sourceViewName)
         sql(s"CREATE TABLE $targetTabName LIKE $sourceViewName")
 
-        val sourceTable = spark.sessionState.catalog.getTempViewMetadataOption(sourceViewName).get
+        val sourceTable = spark.sessionState.catalog.getTempViewOrPermanentTableMetadata(
+          TableIdentifier(sourceViewName))
         val targetTable = spark.sessionState.catalog.getTableMetadata(
           TableIdentifier(targetTabName, Some("default")))
 

From ee3096ca6bc8b3fb8962694169d12c56ca9aa236 Mon Sep 17 00:00:00 2001
From: gatorsmile <gatorsmile@gmail.com>
Date: Mon, 19 Sep 2016 23:15:28 -0700
Subject: [PATCH 10/10] address comments

---
 .../sql/catalyst/catalog/SessionCatalog.scala |  6 ++---
 .../catalog/SessionCatalogSuite.scala         | 25 +++----------------
 .../spark/sql/execution/command/tables.scala  |  4 +--
 .../spark/sql/internal/CatalogImpl.scala      |  2 +-
 .../sql/hive/execution/HiveDDLSuite.scala     |  4 +--
 5 files changed, 11 insertions(+), 30 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index 8c6aa6a913f1..ef29c75c0189 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -277,15 +277,15 @@ class SessionCatalog(
    * If the specified table/view is not found in the database then a [[NoSuchTableException]] is
    * thrown.
    */
-  def getTempViewOrPermanentTableMetadata(name: TableIdentifier): CatalogTable = synchronized {
-    val table = formatTableName(name.table)
+  def getTempViewOrPermanentTableMetadata(name: String): CatalogTable = synchronized {
+    val table = formatTableName(name)
     getTempView(table).map { plan =>
       CatalogTable(
         identifier = TableIdentifier(table),
         tableType = CatalogTableType.VIEW,
         storage = CatalogStorageFormat.empty,
         schema = plan.output.toStructType)
-    }.getOrElse(getTableMetadata(name))
+    }.getOrElse(getTableMetadata(TableIdentifier(name)))
   }
 
   /**
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
index e071ca8e1225..384a7308615e 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
@@ -444,35 +444,16 @@ class SessionCatalogSuite extends SparkFunSuite {
     assert(!catalog.tableExists(TableIdentifier("view1", Some("default"))))
   }
 
-  test("getTableMetadata and getTempViewOrPermanentTableMetadata on temporary views") {
+  test("getTempViewOrPermanentTableMetadata on temporary views") {
     val catalog = new SessionCatalog(newBasicCatalog())
     val tempTable = Range(1, 10, 2, 10)
-    val m = intercept[AnalysisException] {
-      catalog.getTableMetadata(TableIdentifier("view1"))
-    }.getMessage
-    assert(m.contains("Table or view 'view1' not found in database 'default'"))
-
-    val m2 = intercept[AnalysisException] {
-      catalog.getTableMetadata(TableIdentifier("view1", Some("default")))
-    }.getMessage
-    assert(m2.contains("Table or view 'view1' not found in database 'default'"))
-
     intercept[NoSuchTableException] {
-      catalog.getTempViewOrPermanentTableMetadata(TableIdentifier("view1"))
+      catalog.getTempViewOrPermanentTableMetadata("view1")
     }.getMessage
 
     catalog.createTempView("view1", tempTable, overrideIfExists = false)
-    assert(catalog.getTempViewOrPermanentTableMetadata(TableIdentifier("view1")).identifier ==
+    assert(catalog.getTempViewOrPermanentTableMetadata("view1").identifier ==
       TableIdentifier("view1"), "the temporary view `view1` should exist")
-
-    intercept[NoSuchTableException] {
-      catalog.getTableMetadata(TableIdentifier("view1"))
-    }.getMessage
-
-    val m3 = intercept[AnalysisException] {
-      catalog.getTableMetadata(TableIdentifier("view1", Some("default")))
-    }.getMessage
-    assert(m3.contains("Table or view 'view1' not found in database 'default'"))
   }
 
   test("list tables without pattern") {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 0bafdaa184a0..94b46c5d9715 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -67,7 +67,7 @@ case class CreateTableLikeCommand(
     val sourceTableDesc = if (sourceTable.database.isDefined) {
       catalog.getTableMetadata(sourceTable)
     } else {
-      catalog.getTempViewOrPermanentTableMetadata(sourceTable)
+      catalog.getTempViewOrPermanentTableMetadata(sourceTable.table)
     }
 
     // Storage format
@@ -605,7 +605,7 @@ case class ShowColumnsCommand(tableName: TableIdentifier) extends RunnableComman
     val table = if (tableName.database.isDefined) {
       catalog.getTableMetadata(tableName)
     } else {
-      catalog.getTempViewOrPermanentTableMetadata(tableName)
+      catalog.getTempViewOrPermanentTableMetadata(tableName.table)
     }
     table.schema.map { c =>
       Row(c.name)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
index ee757e7dc166..6fecda232ab8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
@@ -154,7 +154,7 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
     val tableMetadata = if (tableIdentifier.database.isDefined) {
       sessionCatalog.getTableMetadata(tableIdentifier)
     } else {
-      sessionCatalog.getTempViewOrPermanentTableMetadata(tableIdentifier)
+      sessionCatalog.getTempViewOrPermanentTableMetadata(tableIdentifier.table)
     }
 
     val partitionColumnNames = tableMetadata.partitionColumnNames.toSet
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index c927e5d802c9..38482f66a38e 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -678,8 +678,8 @@ class HiveDDLSuite
           .createTempView(sourceViewName)
         sql(s"CREATE TABLE $targetTabName LIKE $sourceViewName")
 
-        val sourceTable = spark.sessionState.catalog.getTempViewOrPermanentTableMetadata(
-          TableIdentifier(sourceViewName))
+        val sourceTable =
+          spark.sessionState.catalog.getTempViewOrPermanentTableMetadata(sourceViewName)
         val targetTable = spark.sessionState.catalog.getTableMetadata(
           TableIdentifier(targetTabName, Some("default")))