[SPARK-14119][SPARK-14120][SPARK-14122][SQL] Throw exception on unsupported DDL commands

## What changes were proposed in this pull request?

Before: We just pass all role commands to Hive even though it doesn't work.
After: We throw an `AnalysisException` that looks like this:

```
scala> sql("CREATE ROLE x")
org.apache.spark.sql.AnalysisException: Unsupported Hive operation: CREATE ROLE;
  at org.apache.spark.sql.hive.HiveQl$$anonfun$parsePlan$1.apply(HiveQl.scala:213)
  at org.apache.spark.sql.hive.HiveQl$$anonfun$parsePlan$1.apply(HiveQl.scala:208)
  at org.apache.spark.sql.catalyst.parser.CatalystQl.safeParse(CatalystQl.scala:49)
  at org.apache.spark.sql.hive.HiveQl.parsePlan(HiveQl.scala:208)
  at org.apache.spark.sql.SQLContext.parseSql(SQLContext.scala:198)
```

## How was this patch tested?

`HiveQuerySuite`

Author: Andrew Or <andrew@databricks.com>

Closes #11948 from andrewor14/ddl-role-management.

3 files changed, 69 insertions, 32 deletions

diff --git a/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala b/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
index 650797f768..bedbf9ae17 100644
--- a/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
+++ b/sql/hive/compatibility/src/test/scala/org/apache/spark/sql/hive/execution/HiveCompatibilitySuite.scala
@@ -291,7 +291,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "compute_stats_empty_table",
     "compute_stats_long",
     "create_view_translate",
-    "show_create_table_serde",
     "show_tblproperties",
 
     // Odd changes to output
@@ -344,6 +343,15 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     // These tests check the VIEW table definition, but Spark handles CREATE VIEW itself and
     // generates different View Expanded Text.
     "alter_view_as_select",
+
+    // We don't support show create table commands in general
+    "show_create_table_alter",
+    "show_create_table_db_table",
+    "show_create_table_delimited",
+    "show_create_table_does_not_exist",
+    "show_create_table_index",
+    "show_create_table_partitioned",
+    "show_create_table_serde",
     "show_create_table_view"
   )
 
@@ -833,13 +841,6 @@ class HiveCompatibilitySuite extends HiveQueryFileTest with BeforeAndAfter {
     "serde_reported_schema",
     "set_variable_sub",
     "show_columns",
-    "show_create_table_alter",
-    "show_create_table_db_table",
-    "show_create_table_delimited",
-    "show_create_table_does_not_exist",
-    "show_create_table_index",
-    "show_create_table_partitioned",
-    "show_create_table_serde",
     "show_describe_func_quotes",
     "show_functions",
     "show_partitions",
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveQl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveQl.scala
index b3ec95fc73..052c43a3ce 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveQl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveQl.scala
@@ -31,7 +31,6 @@ import org.apache.hadoop.hive.serde.serdeConstants
 import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe
 
 import org.apache.spark.internal.Logging
-import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.parser._
@@ -83,7 +82,28 @@ private[hive] class HiveQl(conf: ParserConf) extends SparkQl(conf) with Logging
   import ParseUtils._
   import ParserUtils._
 
-  protected val nativeCommands = Seq(
+  // Token text -> human readable text
+  private val hiveUnsupportedCommands = Map(
+    "TOK_CREATEROLE" -> "CREATE ROLE",
+    "TOK_DROPROLE" -> "DROP ROLE",
+    "TOK_EXPORT" -> "EXPORT TABLE",
+    "TOK_GRANT" -> "GRANT",
+    "TOK_GRANT_ROLE" -> "GRANT",
+    "TOK_IMPORT" -> "IMPORT TABLE",
+    "TOK_REVOKE" -> "REVOKE",
+    "TOK_REVOKE_ROLE" -> "REVOKE",
+    "TOK_SHOW_COMPACTIONS" -> "SHOW COMPACTIONS",
+    "TOK_SHOW_CREATETABLE" -> "SHOW CREATE TABLE",
+    "TOK_SHOW_GRANT" -> "SHOW GRANT",
+    "TOK_SHOW_ROLE_GRANT" -> "SHOW ROLE GRANT",
+    "TOK_SHOW_ROLE_PRINCIPALS" -> "SHOW PRINCIPALS",
+    "TOK_SHOW_ROLES" -> "SHOW ROLES",
+    "TOK_SHOW_SET_ROLE" -> "SHOW CURRENT ROLES / SET ROLE",
+    "TOK_SHOW_TRANSACTIONS" -> "SHOW TRANSACTIONS",
+    "TOK_SHOWINDEXES" -> "SHOW INDEXES",
+    "TOK_SHOWLOCKS" -> "SHOW LOCKS")
+
+  private val nativeCommands = Set(
     "TOK_ALTERDATABASE_OWNER",
     "TOK_ALTERINDEX_PROPERTIES",
     "TOK_ALTERINDEX_REBUILD",
@@ -97,51 +117,30 @@ private[hive] class HiveQl(conf: ParserConf) extends SparkQl(conf) with Logging
 
     "TOK_CREATEINDEX",
     "TOK_CREATEMACRO",
-    "TOK_CREATEROLE",
 
     "TOK_DROPINDEX",
     "TOK_DROPMACRO",
-    "TOK_DROPROLE",
     "TOK_DROPTABLE_PROPERTIES",
     "TOK_DROPVIEW",
     "TOK_DROPVIEW_PROPERTIES",
 
-    "TOK_EXPORT",
-
-    "TOK_GRANT",
-    "TOK_GRANT_ROLE",
-
-    "TOK_IMPORT",
-
     "TOK_LOAD",
 
     "TOK_LOCKTABLE",
 
     "TOK_MSCK",
 
-    "TOK_REVOKE",
-
-    "TOK_SHOW_COMPACTIONS",
-    "TOK_SHOW_CREATETABLE",
-    "TOK_SHOW_GRANT",
-    "TOK_SHOW_ROLE_GRANT",
-    "TOK_SHOW_ROLE_PRINCIPALS",
-    "TOK_SHOW_ROLES",
-    "TOK_SHOW_SET_ROLE",
     "TOK_SHOW_TABLESTATUS",
     "TOK_SHOW_TBLPROPERTIES",
-    "TOK_SHOW_TRANSACTIONS",
     "TOK_SHOWCOLUMNS",
     "TOK_SHOWDATABASES",
-    "TOK_SHOWINDEXES",
-    "TOK_SHOWLOCKS",
     "TOK_SHOWPARTITIONS",
 
     "TOK_UNLOCKTABLE"
   )
 
   // Commands that we do not need to explain.
-  protected val noExplainCommands = Seq(
+  private val noExplainCommands = Set(
     "TOK_DESCTABLE",
     "TOK_SHOWTABLES",
     "TOK_TRUNCATETABLE", // truncate table" is a NativeCommand, does not need to explain.
@@ -209,6 +208,9 @@ private[hive] class HiveQl(conf: ParserConf) extends SparkQl(conf) with Logging
     safeParse(sql, ParseDriver.parsePlan(sql, conf)) { ast =>
       if (nativeCommands.contains(ast.text)) {
         HiveNativeCommand(sql)
+      } else if (hiveUnsupportedCommands.contains(ast.text)) {
+        val humanReadableText = hiveUnsupportedCommands(ast.text)
+        throw new AnalysisException("Unsupported operation: " + humanReadableText)
       } else {
         nodeToPlan(ast) match {
           case NativePlaceholder => HiveNativeCommand(sql)
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala
index 197a123905..79774f5913 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala
@@ -68,6 +68,11 @@ class HiveQuerySuite extends HiveComparisonTest with BeforeAndAfter {
     }
   }
 
+  private def assertUnsupportedFeature(body: => Unit): Unit = {
+    val e = intercept[AnalysisException] { body }
+    assert(e.getMessage.toLowerCase.contains("unsupported operation"))
+  }
+
   test("SPARK-4908: concurrent hive native commands") {
     (1 to 100).par.map { _ =>
       sql("USE default")
@@ -1246,6 +1251,35 @@ class HiveQuerySuite extends HiveComparisonTest with BeforeAndAfter {
 
   // Put tests that depend on specific Hive settings before these last two test,
   // since they modify /clear stuff.
+
+  test("role management commands are not supported") {
+    assertUnsupportedFeature { sql("CREATE ROLE my_role") }
+    assertUnsupportedFeature { sql("DROP ROLE my_role") }
+    assertUnsupportedFeature { sql("SHOW CURRENT ROLES") }
+    assertUnsupportedFeature { sql("SHOW ROLES") }
+    assertUnsupportedFeature { sql("SHOW GRANT") }
+    assertUnsupportedFeature { sql("SHOW ROLE GRANT USER my_principal") }
+    assertUnsupportedFeature { sql("SHOW PRINCIPALS my_role") }
+    assertUnsupportedFeature { sql("SET ROLE my_role") }
+    assertUnsupportedFeature { sql("GRANT my_role TO USER my_user") }
+    assertUnsupportedFeature { sql("GRANT ALL ON my_table TO USER my_user") }
+    assertUnsupportedFeature { sql("REVOKE my_role FROM USER my_user") }
+    assertUnsupportedFeature { sql("REVOKE ALL ON my_table FROM USER my_user") }
+  }
+
+  test("import/export commands are not supported") {
+    assertUnsupportedFeature { sql("IMPORT TABLE my_table FROM 'my_path'") }
+    assertUnsupportedFeature { sql("EXPORT TABLE my_table TO 'my_path'") }
+  }
+
+  test("some show commands are not supported") {
+    assertUnsupportedFeature { sql("SHOW CREATE TABLE my_table") }
+    assertUnsupportedFeature { sql("SHOW COMPACTIONS") }
+    assertUnsupportedFeature { sql("SHOW TRANSACTIONS") }
+    assertUnsupportedFeature { sql("SHOW INDEXES ON my_table") }
+    assertUnsupportedFeature { sql("SHOW LOCKS my_table") }
+  }
+
 }
 
 // for SPARK-2180 test