[SPARK-14362][SPARK-14406][SQL] DDL Native Support: Drop View and Drop Table

#### What changes were proposed in this pull request?

This PR is to provide a native support for DDL `DROP VIEW` and `DROP TABLE`. The PR includes native parsing and native analysis.

Based on the HIVE DDL document for [DROP_VIEW_WEB_LINK](https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-
DropView
), `DROP VIEW` is defined as,
**Syntax:**
```SQL
DROP VIEW [IF EXISTS] [db_name.]view_name;
```
 - to remove metadata for the specified view.
 - illegal to use DROP TABLE on a view.
 - illegal to use DROP VIEW on a table.
 - this command only works in `HiveContext`. In `SQLContext`, we will get an exception.

This PR also handles `DROP TABLE`.
**Syntax:**
```SQL
DROP TABLE [IF EXISTS] table_name [PURGE];
```
- Previously, the `DROP TABLE` command only can drop Hive tables in `HiveContext`. Now, after this PR, this command also can drop temporary table, external table, external data source table in `SQLContext`.
- In `HiveContext`, we will not issue an exception if the to-be-dropped table does not exist and users did not specify `IF EXISTS`. Instead, we just log an error message. If `IF EXISTS` is specified, we will not issue any error message/exception.
- In `SQLContext`, we will issue an exception if the to-be-dropped table does not exist, unless `IF EXISTS` is specified.
- Data will not be deleted if the tables are `external`, unless table type is `managed_table`.

#### How was this patch tested?
For verifying command parsing, added test cases in `spark/sql/hive/HiveDDLCommandSuite.scala`
For verifying command analysis, added test cases in `spark/sql/hive/execution/HiveDDLSuite.scala`

Author: gatorsmile <[email protected]>
Author: xiaoli <[email protected]>
Author: Xiao Li <[email protected]>

Closes #12146 from gatorsmile/dropView.

Author: gatorsmile

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

@@ -104,6 +104,7 @@ statement
         REPLACE COLUMNS '(' colTypeList ')' (CASCADE | RESTRICT)?      #replaceColumns
     | DROP TABLE (IF EXISTS)? tableIdentifier PURGE?
         (FOR METADATA? REPLICATION '(' STRING ')')?                    #dropTable
+    | DROP VIEW (IF EXISTS)? tableIdentifier                           #dropTable
     | CREATE (OR REPLACE)? VIEW (IF NOT EXISTS)? tableIdentifier
         identifierCommentList? (COMMENT STRING)?
         (PARTITIONED ON identifierList)?
@@ -141,7 +142,6 @@ hiveNativeCommands
     | DELETE FROM tableIdentifier (WHERE booleanExpression)?
     | TRUNCATE TABLE tableIdentifier partitionSpec?
         (COLUMNS identifierList)?
-    | DROP VIEW (IF EXISTS)? qualifiedName
     | SHOW COLUMNS (FROM | IN) tableIdentifier ((FROM|IN) identifier)?
     | START TRANSACTION (transactionMode (',' transactionMode)*)?
     | COMMIT WORK?

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InMemoryCatalog.scala

@@ -187,6 +187,10 @@ class InMemoryCatalog extends ExternalCatalog {
     catalog(db).tables(table).table
   }
 
+  override def getTableOption(db: String, table: String): Option[CatalogTable] = synchronized {
+    if (!tableExists(db, table)) None else Option(catalog(db).tables(table).table)
+  }
+
   override def tableExists(db: String, table: String): Boolean = synchronized {
     requireDbExists(db)
     catalog(db).tables.contains(table)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -21,6 +21,7 @@ import java.io.File
 
 import scala.collection.mutable
 
+import org.apache.spark.internal.Logging
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.{CatalystConf, SimpleCatalystConf}
 import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
@@ -41,7 +42,7 @@ class SessionCatalog(
     externalCatalog: ExternalCatalog,
     functionResourceLoader: FunctionResourceLoader,
     functionRegistry: FunctionRegistry,
-    conf: CatalystConf) {
+    conf: CatalystConf) extends Logging {
   import ExternalCatalog._
 
   def this(
@@ -175,6 +176,17 @@ class SessionCatalog(
     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.
+   * If the specified table is not found in the database then return None if it doesn't exist.
+   */
+  def getTableMetadataOption(name: TableIdentifier): Option[CatalogTable] = {
+    val db = name.database.getOrElse(currentDb)
+    val table = formatTableName(name.table)
+    externalCatalog.getTableOption(db, table)
+  }
+
   // -------------------------------------------------------------
   // | Methods that interact with temporary and metastore tables |
   // -------------------------------------------------------------
@@ -229,7 +241,13 @@ class SessionCatalog(
     val db = name.database.getOrElse(currentDb)
     val table = formatTableName(name.table)
     if (name.database.isDefined || !tempTables.contains(table)) {
-      externalCatalog.dropTable(db, table, ignoreIfNotExists)
+      // When ignoreIfNotExists is false, no exception is issued when the table does not exist.
+      // Instead, log it as an error message. This is consistent with Hive.
+      if (externalCatalog.tableExists(db, table)) {
+        externalCatalog.dropTable(db, table, ignoreIfNotExists = true)
+      } else if (!ignoreIfNotExists) {
+        logError(s"Table '${name.quotedString}' does not exist")
+      }
     } else {
       tempTables.remove(table)
     }
@@ -283,9 +301,14 @@ class SessionCatalog(
    * explicitly specified.
    */
   def isTemporaryTable(name: TableIdentifier): Boolean = {
-    !name.database.isDefined && tempTables.contains(formatTableName(name.table))
+    name.database.isEmpty && tempTables.contains(formatTableName(name.table))
   }
 
+  /**
+   * Return whether View is supported
+   */
+  def isViewSupported: Boolean = false
+
   /**
    * List all tables in the specified database, including temporary tables.
    */

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala

@@ -91,6 +91,8 @@ abstract class ExternalCatalog {
 
   def getTable(db: String, table: String): CatalogTable
 
+  def getTableOption(db: String, table: String): Option[CatalogTable]
+
   def tableExists(db: String, table: String): Boolean
 
   def listTables(db: String): Seq[String]

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala

@@ -233,10 +233,9 @@ class SessionCatalogSuite extends SparkFunSuite {
     intercept[AnalysisException] {
       catalog.dropTable(TableIdentifier("tbl1", Some("unknown_db")), ignoreIfNotExists = true)
     }
-    // Table does not exist
-    intercept[AnalysisException] {
-      catalog.dropTable(TableIdentifier("unknown_table", Some("db2")), ignoreIfNotExists = false)
-    }
+    // If the table does not exist, we do not issue an exception. Instead, we output an error log
+    // message to console when ignoreIfNotExists is set to false.
+    catalog.dropTable(TableIdentifier("unknown_table", Some("db2")), ignoreIfNotExists = false)
     catalog.dropTable(TableIdentifier("unknown_table", Some("db2")), ignoreIfNotExists = true)
   }
 

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

@@ -363,6 +363,22 @@ class SparkSqlAstBuilder extends AstBuilder {
     }
   }
 
+  /**
+   * Create a [[DropTable]] command.
+   */
+  override def visitDropTable(ctx: DropTableContext): LogicalPlan = withOrigin(ctx) {
+    if (ctx.PURGE != null) {
+      throw new ParseException("Unsupported operation: PURGE option", ctx)
+    }
+    if (ctx.REPLICATION != null) {
+      throw new ParseException("Unsupported operation: REPLICATION clause", ctx)
+    }
+    DropTable(
+      visitTableIdentifier(ctx.tableIdentifier),
+      ctx.EXISTS != null,
+      ctx.VIEW != null)
+  }
+
   /**
    * Create a [[AlterTableRename]] command.
    *

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

@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.command
 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.{CatalogDatabase, CatalogTable, CatalogTableType}
 import org.apache.spark.sql.catalyst.catalog.ExternalCatalog.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
 import org.apache.spark.sql.types._
@@ -175,13 +175,61 @@ case class DescribeDatabase(
   }
 }
 
+/**
+ * Drops a table/view from the metastore and removes it if it is cached.
+ *
+ * The syntax of this command is:
+ * {{{
+ *   DROP TABLE [IF EXISTS] table_name;
+ *   DROP VIEW [IF EXISTS] [db_name.]view_name;
+ * }}}
+ */
+case class DropTable(
+    tableName: TableIdentifier,
+    ifExists: Boolean,
+    isView: Boolean) extends RunnableCommand {
+
+  override def run(sqlContext: SQLContext): Seq[Row] = {
+    val catalog = sqlContext.sessionState.catalog
+    if (isView && !catalog.isViewSupported) {
+      throw new AnalysisException(s"Not supported object: views")
+    }
+    // If the command DROP VIEW is to drop a table or DROP TABLE is to drop a view
+    // issue an exception.
+    catalog.getTableMetadataOption(tableName).map(_.tableType match {
+      case CatalogTableType.VIRTUAL_VIEW if !isView =>
+        throw new AnalysisException(
+          "Cannot drop a view with DROP TABLE. Please use DROP VIEW instead")
+      case o if o != CatalogTableType.VIRTUAL_VIEW && isView =>
+        throw new AnalysisException(
+          s"Cannot drop a table with DROP VIEW. Please use DROP TABLE instead")
+      case _ =>
+    })
+
+    try {
+      sqlContext.cacheManager.tryUncacheQuery(sqlContext.table(tableName.quotedString))
+    } catch {
+      // This table's metadata is not in Hive metastore (e.g. the table does not exist).
+      case e if e.getClass.getName == "org.apache.hadoop.hive.ql.metadata.InvalidTableException" =>
+      case _: org.apache.spark.sql.catalyst.analysis.NoSuchTableException =>
+      // Other Throwables can be caused by users providing wrong parameters in OPTIONS
+      // (e.g. invalid paths). We catch it and log a warning message.
+      // Users should be able to drop such kinds of tables regardless if there is an error.
+      case e: Throwable => log.warn(s"${e.getMessage}", e)
+    }
+    catalog.invalidateTable(tableName)
+    catalog.dropTable(tableName, ifExists)
+    Seq.empty[Row]
+  }
+}
+
 /**
  * A command that renames a table/view.
  *
  * The syntax of this command is:
  * {{{
- *    ALTER TABLE table1 RENAME TO table2;
- *    ALTER VIEW view1 RENAME TO view2;
+ *   ALTER TABLE table1 RENAME TO table2;
+ *   ALTER VIEW view1 RENAME TO view2;
  * }}}
  */
 case class AlterTableRename(

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

@@ -20,9 +20,8 @@ package org.apache.spark.sql.execution.command
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.plans.PlanTest
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.plans.logical.Project
 import org.apache.spark.sql.execution.SparkSqlParser
-import org.apache.spark.sql.execution.datasources.BucketSpec
 import org.apache.spark.sql.types._
 
 class DDLCommandSuite extends PlanTest {
@@ -667,7 +666,10 @@ class DDLCommandSuite extends PlanTest {
 
   test("unsupported operations") {
     intercept[ParseException] {
-      parser.parsePlan("DROP TABLE D1.T1")
+      parser.parsePlan("DROP TABLE tab PURGE")
+    }
+    intercept[ParseException] {
+      parser.parsePlan("DROP TABLE tab FOR REPLICATION('eventid')")
     }
     intercept[ParseException] {
       parser.parsePlan("CREATE VIEW testView AS SELECT id FROM tab")
@@ -700,4 +702,52 @@ class DDLCommandSuite extends PlanTest {
     val parsed = parser.parsePlan(sql)
     assert(parsed.isInstanceOf[Project])
   }
+
+  test("drop table") {
+    val tableName1 = "db.tab"
+    val tableName2 = "tab"
+
+    val parsed1 = parser.parsePlan(s"DROP TABLE $tableName1")
+    val parsed2 = parser.parsePlan(s"DROP TABLE IF EXISTS $tableName1")
+    val parsed3 = parser.parsePlan(s"DROP TABLE $tableName2")
+    val parsed4 = parser.parsePlan(s"DROP TABLE IF EXISTS $tableName2")
+
+    val expected1 =
+      DropTable(TableIdentifier("tab", Option("db")), ifExists = false, isView = false)
+    val expected2 =
+      DropTable(TableIdentifier("tab", Option("db")), ifExists = true, isView = false)
+    val expected3 =
+      DropTable(TableIdentifier("tab", None), ifExists = false, isView = false)
+    val expected4 =
+      DropTable(TableIdentifier("tab", None), ifExists = true, isView = false)
+
+    comparePlans(parsed1, expected1)
+    comparePlans(parsed2, expected2)
+    comparePlans(parsed3, expected3)
+    comparePlans(parsed4, expected4)
+  }
+
+  test("drop view") {
+    val viewName1 = "db.view"
+    val viewName2 = "view"
+
+    val parsed1 = parser.parsePlan(s"DROP VIEW $viewName1")
+    val parsed2 = parser.parsePlan(s"DROP VIEW IF EXISTS $viewName1")
+    val parsed3 = parser.parsePlan(s"DROP VIEW $viewName2")
+    val parsed4 = parser.parsePlan(s"DROP VIEW IF EXISTS $viewName2")
+
+    val expected1 =
+      DropTable(TableIdentifier("view", Option("db")), ifExists = false, isView = true)
+    val expected2 =
+      DropTable(TableIdentifier("view", Option("db")), ifExists = true, isView = true)
+    val expected3 =
+      DropTable(TableIdentifier("view", None), ifExists = false, isView = true)
+    val expected4 =
+      DropTable(TableIdentifier("view", None), ifExists = true, isView = true)
+
+    comparePlans(parsed1, expected1)
+    comparePlans(parsed2, expected2)
+    comparePlans(parsed3, expected3)
+    comparePlans(parsed4, expected4)
+  }
 }

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

@@ -391,6 +391,54 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
       Nil)
   }
 
+  test("drop table - temporary table") {
+    val catalog = sqlContext.sessionState.catalog
+    sql(
+      """
+        |CREATE TEMPORARY TABLE tab1
+        |USING org.apache.spark.sql.sources.DDLScanSource
+        |OPTIONS (
+        |  From '1',
+        |  To '10',
+        |  Table 'test1'
+        |)
+      """.stripMargin)
+    assert(catalog.listTables("default") == Seq(TableIdentifier("tab1")))
+    sql("DROP TABLE tab1")
+    assert(catalog.listTables("default") == Nil)
+  }
+
+  test("drop table") {
+    testDropTable(isDatasourceTable = false)
+  }
+
+  test("drop table - data source table") {
+    testDropTable(isDatasourceTable = true)
+  }
+
+  private def testDropTable(isDatasourceTable: Boolean): Unit = {
+    val catalog = sqlContext.sessionState.catalog
+    val tableIdent = TableIdentifier("tab1", Some("dbx"))
+    createDatabase(catalog, "dbx")
+    createTable(catalog, tableIdent)
+    if (isDatasourceTable) {
+      convertToDatasourceTable(catalog, tableIdent)
+    }
+    assert(catalog.listTables("dbx") == Seq(tableIdent))
+    sql("DROP TABLE dbx.tab1")
+    assert(catalog.listTables("dbx") == Nil)
+    sql("DROP TABLE IF EXISTS dbx.tab1")
+    // no exception will be thrown
+    sql("DROP TABLE dbx.tab1")
+  }
+
+  test("drop view") {
+    val e = intercept[AnalysisException] {
+      sql("DROP VIEW dbx.tab1")
+    }
+    assert(e.getMessage.contains("Not supported object: views"))
+  }
+
   private def convertToDatasourceTable(
       catalog: SessionCatalog,
       tableIdent: TableIdentifier): Unit = {

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala

@@ -172,7 +172,7 @@ class CliSuite extends SparkFunSuite with BeforeAndAfterAll with Logging {
       "SELECT COUNT(*) FROM hive_test;"
         -> "5",
       "DROP TABLE hive_test;"
-        -> "OK"
+        -> ""
     )
   }
 
@@ -220,9 +220,9 @@ class CliSuite extends SparkFunSuite with BeforeAndAfterAll with Logging {
       "SELECT count(key) FROM t1;"
         -> "5",
       "DROP TABLE t1;"
-        -> "OK",
+        -> "",
       "DROP TABLE sourceTable;"
-        -> "OK"
+        -> ""
     )
   }