[SPARK-14125] [SQL] Native DDL Support: Alter View

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -393,7 +393,8 @@ class SparkSqlAstBuilder extends AstBuilder {
   override def visitRenameTable(ctx: RenameTableContext): LogicalPlan = withOrigin(ctx) {
     AlterTableRename(
       visitTableIdentifier(ctx.from),
-      visitTableIdentifier(ctx.to))
+      visitTableIdentifier(ctx.to),
+      ctx.VIEW != null)
   }
 
   /**
@@ -409,7 +410,8 @@ class SparkSqlAstBuilder extends AstBuilder {
       ctx: SetTablePropertiesContext): LogicalPlan = withOrigin(ctx) {
     AlterTableSetProperties(
       visitTableIdentifier(ctx.tableIdentifier),
-      visitTablePropertyList(ctx.tablePropertyList))
+      visitTablePropertyList(ctx.tablePropertyList),
+      ctx.VIEW != null)
   }
 
   /**
@@ -426,7 +428,8 @@ class SparkSqlAstBuilder extends AstBuilder {
     AlterTableUnsetProperties(
       visitTableIdentifier(ctx.tableIdentifier),
       visitTablePropertyList(ctx.tablePropertyList).keys.toSeq,
-      ctx.EXISTS != null)
+      ctx.EXISTS != null,
+      ctx.VIEW != null)
   }
 
   /**

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

@@ -23,7 +23,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{AnalysisException, Row, SQLContext}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable}
-import org.apache.spark.sql.catalyst.catalog.{CatalogTablePartition, CatalogTableType}
+import org.apache.spark.sql.catalyst.catalog.{CatalogTablePartition, CatalogTableType, SessionCatalog}
 import org.apache.spark.sql.catalyst.catalog.ExternalCatalog.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
 import org.apache.spark.sql.types._
@@ -235,11 +235,13 @@ case class DropTable(
  */
 case class AlterTableSetProperties(
     tableName: TableIdentifier,
-    properties: Map[String, String])
+    properties: Map[String, String],
+    isView: Boolean)
   extends RunnableCommand {
 
   override def run(sqlContext: SQLContext): Seq[Row] = {
     val catalog = sqlContext.sessionState.catalog
+    DDLUtils.verifyAlterTableType(catalog, tableName, isView)
     val table = catalog.getTableMetadata(tableName)
     val newProperties = table.properties ++ properties
     if (DDLUtils.isDatasourceTable(newProperties)) {
@@ -265,11 +267,13 @@ case class AlterTableSetProperties(
 case class AlterTableUnsetProperties(
     tableName: TableIdentifier,
     propKeys: Seq[String],
-    ifExists: Boolean)
+    ifExists: Boolean,
+    isView: Boolean)
   extends RunnableCommand {
 
   override def run(sqlContext: SQLContext): Seq[Row] = {
     val catalog = sqlContext.sessionState.catalog
+    DDLUtils.verifyAlterTableType(catalog, tableName, isView)
     val table = catalog.getTableMetadata(tableName)
     if (DDLUtils.isDatasourceTable(table)) {
       throw new AnalysisException(
@@ -513,5 +517,24 @@ private object DDLUtils {
   def isDatasourceTable(table: CatalogTable): Boolean = {
     isDatasourceTable(table.properties)
   }
+
+  /**
+   * If the command ALTER VIEW is to alter a table or ALTER TABLE is to alter a view,
+   * issue an exception [[AnalysisException]].
+   */
+  def verifyAlterTableType(
+      catalog: SessionCatalog,
+      tableIdentifier: TableIdentifier,
+      isView: Boolean): Unit = {
+    catalog.getTableMetadataOption(tableIdentifier).map(_.tableType match {
+      case CatalogTableType.VIRTUAL_VIEW if !isView =>
+        throw new AnalysisException(
+          "Cannot alter a view with ALTER TABLE. Please use ALTER VIEW instead")
+      case o if o != CatalogTableType.VIRTUAL_VIEW && isView =>
+        throw new AnalysisException(
+          s"Cannot alter a table with ALTER VIEW. Please use ALTER TABLE instead")
+      case _ =>
+    })
+  }
 }
 

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

@@ -67,11 +67,13 @@ case class CreateTable(table: CatalogTable, ifNotExists: Boolean) extends Runnab
  */
 case class AlterTableRename(
     oldName: TableIdentifier,
-    newName: TableIdentifier)
+    newName: TableIdentifier,
+    isView: Boolean)
   extends RunnableCommand {
 
   override def run(sqlContext: SQLContext): Seq[Row] = {
     val catalog = sqlContext.sessionState.catalog
+    DDLUtils.verifyAlterTableType(catalog, oldName, isView)
     catalog.invalidateTable(oldName)
     catalog.renameTable(oldName, newName)
     Seq.empty[Row]

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala

@@ -214,10 +214,12 @@ class DDLCommandSuite extends PlanTest {
     val parsed_view = parser.parsePlan(sql_view)
     val expected_table = AlterTableRename(
       TableIdentifier("table_name", None),
-      TableIdentifier("new_table_name", None))
+      TableIdentifier("new_table_name", None),
+      isView = false)
     val expected_view = AlterTableRename(
       TableIdentifier("table_name", None),
-      TableIdentifier("new_table_name", None))
+      TableIdentifier("new_table_name", None),
+      isView = true)
     comparePlans(parsed_table, expected_table)
     comparePlans(parsed_view, expected_view)
   }
@@ -244,14 +246,14 @@ class DDLCommandSuite extends PlanTest {
 
     val tableIdent = TableIdentifier("table_name", None)
     val expected1_table = AlterTableSetProperties(
-      tableIdent, Map("test" -> "test", "comment" -> "new_comment"))
+      tableIdent, Map("test" -> "test", "comment" -> "new_comment"), isView = false)
     val expected2_table = AlterTableUnsetProperties(
-      tableIdent, Seq("comment", "test"), ifExists = false)
+      tableIdent, Seq("comment", "test"), ifExists = false, isView = false)
     val expected3_table = AlterTableUnsetProperties(
-      tableIdent, Seq("comment", "test"), ifExists = true)
-    val expected1_view = expected1_table
-    val expected2_view = expected2_table
-    val expected3_view = expected3_table
+      tableIdent, Seq("comment", "test"), ifExists = true, isView = false)
+    val expected1_view = expected1_table.copy(isView = true)
+    val expected2_view = expected2_table.copy(isView = true)
+    val expected3_view = expected3_table.copy(isView = true)
 
     comparePlans(parsed1_table, expected1_table)
     comparePlans(parsed2_table, expected2_table)

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

@@ -131,6 +131,124 @@ class HiveDDLSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
     }
   }
 
+  test("alter views - rename") {
+    val tabName = "tab1"
+    withTable(tabName) {
+      sqlContext.range(10).write.saveAsTable(tabName)
+      val oldViewName = "view1"
+      val newViewName = "view2"
+      withView(oldViewName, newViewName) {
+        val catalog = hiveContext.sessionState.catalog
+        sql(s"CREATE VIEW $oldViewName AS SELECT * FROM $tabName")
+
+        assert(catalog.tableExists(TableIdentifier(oldViewName)))
+        assert(!catalog.tableExists(TableIdentifier(newViewName)))
+        sql(s"ALTER VIEW $oldViewName RENAME TO $newViewName")
+        assert(!catalog.tableExists(TableIdentifier(oldViewName)))
+        assert(catalog.tableExists(TableIdentifier(newViewName)))
+
+        sql(s"DROP VIEW $newViewName")
+      }
+    }
+  }
+
+  test("alter views - set/unset tblproperties") {
+    val tabName = "tab1"
+    withTable(tabName) {
+      sqlContext.range(10).write.saveAsTable(tabName)
+      val viewName = "view1"
+      withView(viewName) {
+        val catalog = hiveContext.sessionState.catalog
+        sql(s"CREATE VIEW $viewName AS SELECT * FROM $tabName")
+
+        assert(catalog.getTableMetadata(TableIdentifier(viewName))
+          .properties.filter(_._1 != "transient_lastDdlTime") == Map())
+        sql(s"ALTER VIEW $viewName SET TBLPROPERTIES ('p' = 'an')")
+        assert(catalog.getTableMetadata(TableIdentifier(viewName))
+          .properties.filter(_._1 != "transient_lastDdlTime") == Map("p" -> "an"))
+
+        // no exception or message will be issued if we set it again
+        sql(s"ALTER VIEW $viewName SET TBLPROPERTIES ('p' = 'an')")
+        assert(catalog.getTableMetadata(TableIdentifier(viewName))
+          .properties.filter(_._1 != "transient_lastDdlTime") == Map("p" -> "an"))
+
+        // the value will be updated if we set the same key to a different value
+        sql(s"ALTER VIEW $viewName SET TBLPROPERTIES ('p' = 'b')")
+        assert(catalog.getTableMetadata(TableIdentifier(viewName))
+          .properties.filter(_._1 != "transient_lastDdlTime") == Map("p" -> "b"))
+
+        sql(s"ALTER VIEW $viewName UNSET TBLPROPERTIES ('p')")
+        assert(catalog.getTableMetadata(TableIdentifier(viewName))
+          .properties.filter(_._1 != "transient_lastDdlTime") == Map())
+
+        val message = intercept[AnalysisException] {
+          sql(s"ALTER VIEW $viewName UNSET TBLPROPERTIES ('p')")
+        }.getMessage
+        assert(message.contains(
+          "attempted to unset non-existent property 'p' in table '`view1`'"))
+
+        sql(s"DROP VIEW $viewName")
+      }
+    }
+  }
+
+  test("alter views and alter table - misuse") {
+    val tabName = "tab1"
+    withTable(tabName) {
+      sqlContext.range(10).write.saveAsTable(tabName)
+      val oldViewName = "view1"
+      val newViewName = "view2"
+      withView(oldViewName, newViewName) {
+        val catalog = hiveContext.sessionState.catalog
+        sql(s"CREATE VIEW $oldViewName AS SELECT * FROM $tabName")
+
+        assert(catalog.tableExists(TableIdentifier(tabName)))
+        assert(catalog.tableExists(TableIdentifier(oldViewName)))
+
+        var message = intercept[AnalysisException] {
+          sql(s"ALTER VIEW $tabName RENAME TO $newViewName")
+        }.getMessage
+        assert(message.contains(
+          "Cannot alter a table with ALTER VIEW. Please use ALTER TABLE instead"))
+
+        message = intercept[AnalysisException] {
+          sql(s"ALTER VIEW $tabName SET TBLPROPERTIES ('p' = 'an')")
+        }.getMessage
+        assert(message.contains(
+          "Cannot alter a table with ALTER VIEW. Please use ALTER TABLE instead"))
+
+        message = intercept[AnalysisException] {
+          sql(s"ALTER VIEW $tabName UNSET TBLPROPERTIES ('p')")
+        }.getMessage
+        assert(message.contains(
+          "Cannot alter a table with ALTER VIEW. Please use ALTER TABLE instead"))
+
+        message = intercept[AnalysisException] {
+          sql(s"ALTER TABLE $oldViewName RENAME TO $newViewName")
+        }.getMessage
+        assert(message.contains(
+          "Cannot alter a view with ALTER TABLE. Please use ALTER VIEW instead"))
+
+        message = intercept[AnalysisException] {
+          sql(s"ALTER TABLE $oldViewName SET TBLPROPERTIES ('p' = 'an')")
+        }.getMessage
+        assert(message.contains(
+          "Cannot alter a view with ALTER TABLE. Please use ALTER VIEW instead"))
+
+        message = intercept[AnalysisException] {
+          sql(s"ALTER TABLE $oldViewName UNSET TBLPROPERTIES ('p')")
+        }.getMessage
+        assert(message.contains(
+          "Cannot alter a view with ALTER TABLE. Please use ALTER VIEW instead"))
+
+        assert(catalog.tableExists(TableIdentifier(tabName)))
+        assert(catalog.tableExists(TableIdentifier(oldViewName)))
+
+        sql(s"DROP VIEW $newViewName")
+      }
+    }
+  }
+
   test("drop table using drop view") {
     withTable("tab1") {
       sql("CREATE TABLE tab1(c1 int)")