[SPARK-17502] [SQL] Fix Multiple Bugs in DDL Statements on Temporary Views

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)
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala

@@ -22,6 +22,7 @@ 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.Statistics
@@ -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 =>

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala

@@ -265,7 +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 (!ifExists) {
       propKeys.foreach { k =>
         if (!table.properties.contains(k)) {

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")
@@ -645,7 +645,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)
@@ -703,7 +703,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)) {

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 TABLE show1a
+          |CREATE TEMPORARY VIEW 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 TABLE show2b
+          |CREATE TEMPORARY VIEW 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 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',
@@ -1605,7 +1605,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")

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")

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,38 @@ 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`"))
+
+      intercept[NoSuchTableException] {
+        sql(s"ANALYZE TABLE $viewName COMPUTE STATISTICS")
+      }
     }
   }
 
+  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) {