[SPARK-14182][SQL] Parse DDL Command: Alter View

This PR is to provide native parsing support for DDL commands: `Alter View`. Since its AST trees are highly similar to `Alter Table`. Thus, both implementation are integrated into the same one.

Based on the Hive DDL document:
https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL and https://cwiki.apache.org/confluence/display/Hive/PartitionedViews

**Syntax:**
```SQL
ALTER VIEW view_name RENAME TO new_view_name
```
 - to change the name of a view to a different name

**Syntax:**
```SQL
ALTER VIEW view_name SET TBLPROPERTIES ('comment' = new_comment);
```
 - to add metadata to a view

**Syntax:**
```SQL
ALTER VIEW view_name UNSET TBLPROPERTIES [IF EXISTS] ('comment', 'key')
```
 - to remove metadata from a view

**Syntax:**
```SQL
ALTER VIEW view_name ADD [IF NOT EXISTS] PARTITION spec1[, PARTITION spec2, ...]
```
 - to add the partitioning metadata for a view.
 - the syntax of partition spec in `ALTER VIEW` is identical to `ALTER TABLE`, **EXCEPT** that it is **ILLEGAL** to specify a `LOCATION` clause.

**Syntax:**
```SQL
ALTER VIEW view_name DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...]
```
 - to drop the related partition metadata for a view.

Added the related test cases to `DDLCommandSuite`

Author: gatorsmile <gatorsmile@gmail.com>
Author: xiaoli <lixiao1983@gmail.com>
Author: Xiao Li <xiaoli@Xiaos-MacBook-Pro.local>

Closes #11987 from gatorsmile/parseAlterView.

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -57,10 +57,11 @@ statement
         (AS? query)?                                                   #createTable
     | ANALYZE TABLE tableIdentifier partitionSpec? COMPUTE STATISTICS
         (identifier | FOR COLUMNS identifierSeq?)?                     #analyze
-    | ALTER TABLE from=tableIdentifier RENAME TO to=tableIdentifier    #renameTable
-    | ALTER TABLE tableIdentifier
+    | ALTER (TABLE | VIEW) from=tableIdentifier
+        RENAME TO to=tableIdentifier                                   #renameTable
+    | ALTER (TABLE | VIEW) tableIdentifier
         SET TBLPROPERTIES tablePropertyList                            #setTableProperties
-    | ALTER TABLE tableIdentifier
+    | ALTER (TABLE | VIEW) tableIdentifier
         UNSET TBLPROPERTIES (IF EXISTS)? tablePropertyList             #unsetTableProperties
     | ALTER TABLE tableIdentifier (partitionSpec)?
         SET SERDE STRING (WITH SERDEPROPERTIES tablePropertyList)?     #setTableSerDe
@@ -76,12 +77,16 @@ statement
         SET SKEWED LOCATION skewedLocationList                         #setTableSkewLocations
     | ALTER TABLE tableIdentifier ADD (IF NOT EXISTS)?
         partitionSpecLocation+                                         #addTablePartition
+    | ALTER VIEW tableIdentifier ADD (IF NOT EXISTS)?
+        partitionSpec+                                                 #addTablePartition
     | ALTER TABLE tableIdentifier
         from=partitionSpec RENAME TO to=partitionSpec                  #renameTablePartition
     | ALTER TABLE from=tableIdentifier
         EXCHANGE partitionSpec WITH TABLE to=tableIdentifier           #exchangeTablePartition
     | ALTER TABLE tableIdentifier
         DROP (IF EXISTS)? partitionSpec (',' partitionSpec)* PURGE?    #dropTablePartitions
+    | ALTER VIEW tableIdentifier
+        DROP (IF EXISTS)? partitionSpec (',' partitionSpec)*           #dropTablePartitions
     | ALTER TABLE tableIdentifier ARCHIVE partitionSpec                #archiveTablePartition
     | ALTER TABLE tableIdentifier UNARCHIVE partitionSpec              #unarchiveTablePartition
     | ALTER TABLE tableIdentifier partitionSpec?
@@ -133,15 +138,6 @@ hiveNativeCommands
     | DELETE FROM tableIdentifier (WHERE booleanExpression)?
     | TRUNCATE TABLE tableIdentifier partitionSpec?
         (COLUMNS identifierList)?
-    | ALTER VIEW from=tableIdentifier AS? RENAME TO to=tableIdentifier
-    | ALTER VIEW from=tableIdentifier AS?
-        SET TBLPROPERTIES tablePropertyList
-    | ALTER VIEW from=tableIdentifier AS?
-        UNSET TBLPROPERTIES (IF EXISTS)? tablePropertyList
-    | ALTER VIEW from=tableIdentifier AS?
-        ADD (IF NOT EXISTS)? partitionSpecLocation+
-    | ALTER VIEW from=tableIdentifier AS?
-        DROP (IF EXISTS)? partitionSpec (',' partitionSpec)* PURGE?
     | DROP VIEW (IF EXISTS)? qualifiedName
     | SHOW COLUMNS (FROM | IN) tableIdentifier ((FROM|IN) identifier)?
     | START TRANSACTION (transactionMode (',' transactionMode)*)?

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

@@ -335,6 +335,7 @@ class SparkSqlAstBuilder extends AstBuilder {
    * For example:
    * {{{
    *   ALTER TABLE table1 RENAME TO table2;
+   *   ALTER VIEW view1 RENAME TO view2;
    * }}}
    */
   override def visitRenameTable(ctx: RenameTableContext): LogicalPlan = withOrigin(ctx) {
@@ -350,6 +351,7 @@ class SparkSqlAstBuilder extends AstBuilder {
    * For example:
    * {{{
    *   ALTER TABLE table SET TBLPROPERTIES ('comment' = new_comment);
+   *   ALTER VIEW view SET TBLPROPERTIES ('comment' = new_comment);
    * }}}
    */
   override def visitSetTableProperties(
@@ -366,6 +368,7 @@ class SparkSqlAstBuilder extends AstBuilder {
    * For example:
    * {{{
    *   ALTER TABLE table UNSET TBLPROPERTIES IF EXISTS ('comment', 'key');
+   *   ALTER VIEW view UNSET TBLPROPERTIES IF EXISTS ('comment', 'key');
    * }}}
    */
   override def visitUnsetTableProperties(
@@ -510,16 +513,22 @@ class SparkSqlAstBuilder extends AstBuilder {
    * For example:
    * {{{
    *   ALTER TABLE table ADD [IF NOT EXISTS] PARTITION spec [LOCATION 'loc1']
+   *   ALTER VIEW view ADD [IF NOT EXISTS] PARTITION spec
    * }}}
    */
   override def visitAddTablePartition(
       ctx: AddTablePartitionContext): LogicalPlan = withOrigin(ctx) {
     // Create partition spec to location mapping.
-    val specsAndLocs = ctx.partitionSpecLocation.asScala.map {
-      splCtx =>
-        val spec = visitNonOptionalPartitionSpec(splCtx.partitionSpec)
-        val location = Option(splCtx.locationSpec).map(visitLocationSpec)
-        spec -> location
+    val specsAndLocs = if (ctx.partitionSpec.isEmpty) {
+      ctx.partitionSpecLocation.asScala.map {
+        splCtx =>
+          val spec = visitNonOptionalPartitionSpec(splCtx.partitionSpec)
+          val location = Option(splCtx.locationSpec).map(visitLocationSpec)
+          spec -> location
+      }
+    } else {
+      // Alter View: the location clauses are not allowed.
+      ctx.partitionSpec.asScala.map(visitNonOptionalPartitionSpec(_) -> None)
     }
     AlterTableAddPartition(
       visitTableIdentifier(ctx.tableIdentifier),
@@ -568,6 +577,7 @@ class SparkSqlAstBuilder extends AstBuilder {
    * For example:
    * {{{
    *   ALTER TABLE table DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...] [PURGE];
+   *   ALTER VIEW view DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...];
    * }}}
    */
   override def visitDropTablePartitions(

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

@@ -195,16 +195,19 @@ case class DropFunction(
     isTemp: Boolean)(sql: String)
   extends NativeDDLCommand(sql) with Logging
 
+/** Rename in ALTER TABLE/VIEW: change the name of a table/view to a different name. */
 case class AlterTableRename(
     oldName: TableIdentifier,
     newName: TableIdentifier)(sql: String)
   extends NativeDDLCommand(sql) with Logging
 
+/** Set Properties in ALTER TABLE/VIEW: add metadata to a table/view. */
 case class AlterTableSetProperties(
     tableName: TableIdentifier,
     properties: Map[String, String])(sql: String)
   extends NativeDDLCommand(sql) with Logging
 
+/** Unset Properties in ALTER TABLE/VIEW: remove metadata from a table/view. */
 case class AlterTableUnsetProperties(
     tableName: TableIdentifier,
     properties: Map[String, String],
@@ -253,6 +256,12 @@ case class AlterTableSkewedLocation(
     skewedMap: Map[String, String])(sql: String)
   extends NativeDDLCommand(sql) with Logging
 
+/**
+ * Add Partition in ALTER TABLE/VIEW: add the table/view partitions.
+ * 'partitionSpecsAndLocs': the syntax of ALTER VIEW is identical to ALTER TABLE,
+ * EXCEPT that it is ILLEGAL to specify a LOCATION clause.
+ * An error message will be issued if the partition exists, unless 'ifNotExists' is true.
+ */
 case class AlterTableAddPartition(
     tableName: TableIdentifier,
     partitionSpecsAndLocs: Seq[(TablePartitionSpec, Option[String])],
@@ -271,6 +280,14 @@ case class AlterTableExchangePartition(
     spec: TablePartitionSpec)(sql: String)
   extends NativeDDLCommand(sql) with Logging
 
+/**
+ * Drop Partition in ALTER TABLE/VIEW: to drop a particular partition for a table/view.
+ * This removes the data and metadata for this partition.
+ * The data is actually moved to the .Trash/Current directory if Trash is configured,
+ * unless 'purge' is true, but the metadata is completely lost.
+ * An error message will be issued if the partition does not exist, unless 'ifExists' is true.
+ * Note: purge is always false when the target is a view.
+ */
 case class AlterTableDropPartition(
     tableName: TableIdentifier,
     specs: Seq[TablePartitionSpec],

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

@@ -195,33 +195,60 @@ class DDLCommandSuite extends PlanTest {
     comparePlans(parsed4, expected4)
   }
 
-  test("alter table: rename table") {
-    val sql = "ALTER TABLE table_name RENAME TO new_table_name"
-    val parsed = parser.parsePlan(sql)
-    val expected = AlterTableRename(
+  // ALTER TABLE table_name RENAME TO new_table_name;
+  // ALTER VIEW view_name RENAME TO new_view_name;
+  test("alter table/view: rename table/view") {
+    val sql_table = "ALTER TABLE table_name RENAME TO new_table_name"
+    val sql_view = sql_table.replace("TABLE", "VIEW")
+    val parsed_table = parser.parsePlan(sql_table)
+    val parsed_view = parser.parsePlan(sql_view)
+    val expected_table = AlterTableRename(
       TableIdentifier("table_name", None),
-      TableIdentifier("new_table_name", None))(sql)
-    comparePlans(parsed, expected)
+      TableIdentifier("new_table_name", None))(sql_table)
+    val expected_view = AlterTableRename(
+      TableIdentifier("table_name", None),
+      TableIdentifier("new_table_name", None))(sql_view)
+    comparePlans(parsed_table, expected_table)
+    comparePlans(parsed_view, expected_view)
   }
 
-  test("alter table: alter table properties") {
-    val sql1 = "ALTER TABLE table_name SET TBLPROPERTIES ('test' = 'test', " +
+  // ALTER TABLE table_name SET TBLPROPERTIES ('comment' = new_comment);
+  // ALTER TABLE table_name UNSET TBLPROPERTIES [IF EXISTS] ('comment', 'key');
+  // ALTER VIEW view_name SET TBLPROPERTIES ('comment' = new_comment);
+  // ALTER VIEW view_name UNSET TBLPROPERTIES [IF EXISTS] ('comment', 'key');
+  test("alter table/view: alter table/view properties") {
+    val sql1_table = "ALTER TABLE table_name SET TBLPROPERTIES ('test' = 'test', " +
       "'comment' = 'new_comment')"
-    val sql2 = "ALTER TABLE table_name UNSET TBLPROPERTIES ('comment', 'test')"
-    val sql3 = "ALTER TABLE table_name UNSET TBLPROPERTIES IF EXISTS ('comment', 'test')"
-    val parsed1 = parser.parsePlan(sql1)
-    val parsed2 = parser.parsePlan(sql2)
-    val parsed3 = parser.parsePlan(sql3)
+    val sql2_table = "ALTER TABLE table_name UNSET TBLPROPERTIES ('comment', 'test')"
+    val sql3_table = "ALTER TABLE table_name UNSET TBLPROPERTIES IF EXISTS ('comment', 'test')"
+    val sql1_view = sql1_table.replace("TABLE", "VIEW")
+    val sql2_view = sql2_table.replace("TABLE", "VIEW")
+    val sql3_view = sql3_table.replace("TABLE", "VIEW")
+
+    val parsed1_table = parser.parsePlan(sql1_table)
+    val parsed2_table = parser.parsePlan(sql2_table)
+    val parsed3_table = parser.parsePlan(sql3_table)
+    val parsed1_view = parser.parsePlan(sql1_view)
+    val parsed2_view = parser.parsePlan(sql2_view)
+    val parsed3_view = parser.parsePlan(sql3_view)
+
     val tableIdent = TableIdentifier("table_name", None)
-    val expected1 = AlterTableSetProperties(
-      tableIdent, Map("test" -> "test", "comment" -> "new_comment"))(sql1)
-    val expected2 = AlterTableUnsetProperties(
-      tableIdent, Map("comment" -> null, "test" -> null), ifExists = false)(sql2)
-    val expected3 = AlterTableUnsetProperties(
-      tableIdent, Map("comment" -> null, "test" -> null), ifExists = true)(sql3)
-    comparePlans(parsed1, expected1)
-    comparePlans(parsed2, expected2)
-    comparePlans(parsed3, expected3)
+    val expected1_table = AlterTableSetProperties(
+      tableIdent, Map("test" -> "test", "comment" -> "new_comment"))(sql1_table)
+    val expected2_table = AlterTableUnsetProperties(
+      tableIdent, Map("comment" -> null, "test" -> null), ifExists = false)(sql2_table)
+    val expected3_table = AlterTableUnsetProperties(
+      tableIdent, Map("comment" -> null, "test" -> null), ifExists = true)(sql3_table)
+    val expected1_view = expected1_table.copy()(sql = sql1_view)
+    val expected2_view = expected2_table.copy()(sql = sql2_view)
+    val expected3_view = expected3_table.copy()(sql = sql3_view)
+
+    comparePlans(parsed1_table, expected1_table)
+    comparePlans(parsed2_table, expected2_table)
+    comparePlans(parsed3_table, expected3_table)
+    comparePlans(parsed1_view, expected1_view)
+    comparePlans(parsed2_view, expected2_view)
+    comparePlans(parsed3_view, expected3_view)
   }
 
   test("alter table: SerDe properties") {
@@ -376,21 +403,66 @@ class DDLCommandSuite extends PlanTest {
     comparePlans(parsed2, expected2)
   }
 
+  // ALTER TABLE table_name ADD [IF NOT EXISTS] PARTITION partition_spec
+  // [LOCATION 'location1'] partition_spec [LOCATION 'location2'] ...;
   test("alter table: add partition") {
-    val sql =
+    val sql1 =
       """
        |ALTER TABLE table_name ADD IF NOT EXISTS PARTITION
        |(dt='2008-08-08', country='us') LOCATION 'location1' PARTITION
        |(dt='2009-09-09', country='uk')
       """.stripMargin
-    val parsed = parser.parsePlan(sql)
-    val expected = AlterTableAddPartition(
+    val sql2 = "ALTER TABLE table_name ADD PARTITION (dt='2008-08-08') LOCATION 'loc'"
+
+    val parsed1 = parser.parsePlan(sql1)
+    val parsed2 = parser.parsePlan(sql2)
+
+    val expected1 = AlterTableAddPartition(
       TableIdentifier("table_name", None),
       Seq(
         (Map("dt" -> "2008-08-08", "country" -> "us"), Some("location1")),
         (Map("dt" -> "2009-09-09", "country" -> "uk"), None)),
-      ifNotExists = true)(sql)
-    comparePlans(parsed, expected)
+      ifNotExists = true)(sql1)
+    val expected2 = AlterTableAddPartition(
+      TableIdentifier("table_name", None),
+      Seq((Map("dt" -> "2008-08-08"), Some("loc"))),
+      ifNotExists = false)(sql2)
+
+    comparePlans(parsed1, expected1)
+    comparePlans(parsed2, expected2)
+  }
+
+  // ALTER VIEW view_name ADD [IF NOT EXISTS] PARTITION partition_spec PARTITION partition_spec ...;
+  test("alter view: add partition") {
+    val sql1 =
+      """
+        |ALTER VIEW view_name ADD IF NOT EXISTS PARTITION
+        |(dt='2008-08-08', country='us') PARTITION
+        |(dt='2009-09-09', country='uk')
+      """.stripMargin
+    // different constant types in partitioning spec
+    val sql2 =
+    """
+      |ALTER VIEW view_name ADD PARTITION
+      |(col1=NULL, cOL2='f', col3=5, COL4=true)
+    """.stripMargin
+
+    val parsed1 = parser.parsePlan(sql1)
+    val parsed2 = parser.parsePlan(sql2)
+
+    val expected1 = AlterTableAddPartition(
+      TableIdentifier("view_name", None),
+      Seq(
+        (Map("dt" -> "2008-08-08", "country" -> "us"), None),
+        (Map("dt" -> "2009-09-09", "country" -> "uk"), None)),
+      ifNotExists = true)(sql1)
+    val expected2 = AlterTableAddPartition(
+      TableIdentifier("view_name", None),
+      Seq((Map("col1" -> "NULL", "col2" -> "f", "col3" -> "5", "col4" -> "true"), None)),
+      ifNotExists = false)(sql2)
+
+    comparePlans(parsed1, expected1)
+    comparePlans(parsed2, expected2)
   }
 
   test("alter table: rename partition") {
@@ -421,36 +493,63 @@ class DDLCommandSuite extends PlanTest {
     comparePlans(parsed, expected)
   }
 
-  test("alter table: drop partitions") {
-    val sql1 =
+  // ALTER TABLE table_name DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...] [PURGE]
+  // ALTER VIEW table_name DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...]
+  test("alter table/view: drop partitions") {
+    val sql1_table =
       """
        |ALTER TABLE table_name DROP IF EXISTS PARTITION
        |(dt='2008-08-08', country='us'), PARTITION (dt='2009-09-09', country='uk')
       """.stripMargin
-    val sql2 =
+    val sql2_table =
       """
        |ALTER TABLE table_name DROP PARTITION
        |(dt='2008-08-08', country='us'), PARTITION (dt='2009-09-09', country='uk') PURGE
       """.stripMargin
-    val parsed1 = parser.parsePlan(sql1)
-    val parsed2 = parser.parsePlan(sql2)
+    val sql1_view = sql1_table.replace("TABLE", "VIEW")
+    // Note: ALTER VIEW DROP PARTITION does not support PURGE
+    val sql2_view = sql2_table.replace("TABLE", "VIEW").replace("PURGE", "")
+
+    val parsed1_table = parser.parsePlan(sql1_table)
+    val parsed2_table = parser.parsePlan(sql2_table)
+    val parsed1_view = parser.parsePlan(sql1_view)
+    val parsed2_view = parser.parsePlan(sql2_view)
+
     val tableIdent = TableIdentifier("table_name", None)
-    val expected1 = AlterTableDropPartition(
+    val expected1_table = AlterTableDropPartition(
       tableIdent,
       Seq(
         Map("dt" -> "2008-08-08", "country" -> "us"),
         Map("dt" -> "2009-09-09", "country" -> "uk")),
       ifExists = true,
-      purge = false)(sql1)
-    val expected2 = AlterTableDropPartition(
+      purge = false)(sql1_table)
+    val expected2_table = AlterTableDropPartition(
       tableIdent,
       Seq(
         Map("dt" -> "2008-08-08", "country" -> "us"),
         Map("dt" -> "2009-09-09", "country" -> "uk")),
       ifExists = false,
-      purge = true)(sql2)
-    comparePlans(parsed1, expected1)
-    comparePlans(parsed2, expected2)
+      purge = true)(sql2_table)
+
+    val expected1_view = AlterTableDropPartition(
+      tableIdent,
+      Seq(
+        Map("dt" -> "2008-08-08", "country" -> "us"),
+        Map("dt" -> "2009-09-09", "country" -> "uk")),
+      ifExists = true,
+      purge = false)(sql1_view)
+    val expected2_view = AlterTableDropPartition(
+      tableIdent,
+      Seq(
+        Map("dt" -> "2008-08-08", "country" -> "us"),
+        Map("dt" -> "2009-09-09", "country" -> "uk")),
+      ifExists = false,
+      purge = false)(sql2_table)
+
+    comparePlans(parsed1_table, expected1_table)
+    comparePlans(parsed2_table, expected2_table)
+    comparePlans(parsed1_view, expected1_view)
+    comparePlans(parsed2_view, expected2_view)
   }
 
   test("alter table: archive partition") {