[SPARK-14184][SQL] Support native execution of SHOW DATABASE command and fix SHOW TABLE to use table identifier pattern

## What changes were proposed in this pull request?

This PR addresses the following

1. Supports native execution of SHOW DATABASES command
2. Fixes SHOW TABLES to apply the identifier_with_wildcards pattern if supplied.

SHOW TABLE syntax
```
SHOW TABLES [IN database_name] ['identifier_with_wildcards'];
```
SHOW DATABASES syntax
```
SHOW (DATABASES|SCHEMAS) [LIKE 'identifier_with_wildcards'];
```

## How was this patch tested?
Tests added in SQLQuerySuite (both hive and sql contexts) and DDLCommandSuite

Note: Since the table name pattern was not working , tests are added in both SQLQuerySuite to
verify the application of the table pattern.

Author: Dilip Biswal <dbiswal@us.ibm.com>

Closes #11991 from dilipbiswal/dkb_show_database.

8 files changed, 163 insertions, 18 deletions

diff --git a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
index a857e670da..5513bbdc7f 100644
--- a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
+++ b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
@@ -114,7 +114,8 @@ statement
     | DROP TEMPORARY? FUNCTION (IF EXISTS)? qualifiedName              #dropFunction
     | EXPLAIN explainOption* statement                                 #explain
     | SHOW TABLES ((FROM | IN) db=identifier)?
-        (LIKE (qualifiedName | pattern=STRING))?                       #showTables
+        (LIKE? pattern=STRING)?                                        #showTables
+    | SHOW DATABASES (LIKE pattern=STRING)?                            #showDatabases
     | SHOW FUNCTIONS (LIKE? (qualifiedName | pattern=STRING))?         #showFunctions
     | (DESC | DESCRIBE) FUNCTION EXTENDED? qualifiedName               #describeFunction
     | (DESC | DESCRIBE) option=(EXTENDED | FORMATTED)?
@@ -618,7 +619,7 @@ number
     ;
 
 nonReserved
-    : SHOW | TABLES | COLUMNS | COLUMN | PARTITIONS | FUNCTIONS
+    : SHOW | TABLES | COLUMNS | COLUMN | PARTITIONS | FUNCTIONS | DATABASES
     | ADD
     | OVER | PARTITION | RANGE | ROWS | PRECEDING | FOLLOWING | CURRENT | ROW | MAP | ARRAY | STRUCT
     | LATERAL | WINDOW | REDUCE | TRANSFORM | USING | SERDE | SERDEPROPERTIES | RECORDREADER
@@ -836,6 +837,7 @@ OUTPUTFORMAT: 'OUTPUTFORMAT';
 INPUTDRIVER: 'INPUTDRIVER';
 OUTPUTDRIVER: 'OUTPUTDRIVER';
 DATABASE: 'DATABASE' | 'SCHEMA';
+DATABASES: 'DATABASES' | 'SCHEMAS';
 DFS: 'DFS';
 TRUNCATE: 'TRUNCATE';
 METADATA: 'METADATA';
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala b/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala
index 0576a1a178..221782ee8f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala
@@ -781,7 +781,7 @@ class SQLContext private[sql](
    * @since 1.3.0
    */
   def tables(): DataFrame = {
-    Dataset.ofRows(this, ShowTablesCommand(None))
+    Dataset.ofRows(this, ShowTablesCommand(None, None))
   }
 
   /**
@@ -793,7 +793,7 @@ class SQLContext private[sql](
    * @since 1.3.0
    */
   def tables(databaseName: String): DataFrame = {
-    Dataset.ofRows(this, ShowTablesCommand(Some(databaseName)))
+    Dataset.ofRows(this, ShowTablesCommand(Some(databaseName), None))
   }
 
   /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 16a899e01f..7efe98dd18 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -70,12 +70,26 @@ class SparkSqlAstBuilder extends AstBuilder {
 
   /**
    * Create a [[ShowTablesCommand]] logical plan.
+   * Example SQL :
+   * {{{
+   *   SHOW TABLES [(IN|FROM) database_name] [[LIKE] 'identifier_with_wildcards'];
+   * }}}
    */
   override def visitShowTables(ctx: ShowTablesContext): LogicalPlan = withOrigin(ctx) {
-    if (ctx.LIKE != null) {
-      logWarning("SHOW TABLES LIKE option is ignored.")
-    }
-    ShowTablesCommand(Option(ctx.db).map(_.getText))
+    ShowTablesCommand(
+      Option(ctx.db).map(_.getText),
+      Option(ctx.pattern).map(string))
+  }
+
+  /**
+   * Create a [[ShowDatabasesCommand]] logical plan.
+   * Example SQL:
+   * {{{
+   *   SHOW (DATABASES|SCHEMAS) [LIKE 'identifier_with_wildcards'];
+   * }}}
+   */
+  override def visitShowDatabases(ctx: ShowDatabasesContext): LogicalPlan = withOrigin(ctx) {
+    ShowDatabasesCommand(Option(ctx.pattern).map(string))
   }
 
   /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala
index 964f0a7a7b..f90d8717ca 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala
@@ -322,18 +322,17 @@ case class DescribeCommand(
  * If a databaseName is not given, the current database will be used.
  * The syntax of using this command in SQL is:
  * {{{
- *    SHOW TABLES [IN databaseName]
+ *   SHOW TABLES [(IN|FROM) database_name] [[LIKE] 'identifier_with_wildcards'];
  * }}}
  */
-case class ShowTablesCommand(databaseName: Option[String]) extends RunnableCommand {
+case class ShowTablesCommand(
+    databaseName: Option[String],
+    tableIdentifierPattern: Option[String]) extends RunnableCommand {
 
   // The result of SHOW TABLES has two columns, tableName and isTemporary.
   override val output: Seq[Attribute] = {
-    val schema = StructType(
-      StructField("tableName", StringType, false) ::
-      StructField("isTemporary", BooleanType, false) :: Nil)
-
-    schema.toAttributes
+    AttributeReference("tableName", StringType, nullable = false)() ::
+      AttributeReference("isTemporary", BooleanType, nullable = false)() :: Nil
   }
 
   override def run(sqlContext: SQLContext): Seq[Row] = {
@@ -341,11 +340,36 @@ case class ShowTablesCommand(databaseName: Option[String]) extends RunnableComma
     // instead of calling tables in sqlContext.
     val catalog = sqlContext.sessionState.catalog
     val db = databaseName.getOrElse(catalog.getCurrentDatabase)
-    val rows = catalog.listTables(db).map { t =>
+    val tables =
+      tableIdentifierPattern.map(catalog.listTables(db, _)).getOrElse(catalog.listTables(db))
+    tables.map { t =>
       val isTemp = t.database.isEmpty
       Row(t.table, isTemp)
     }
-    rows
+  }
+}
+
+/**
+ * A command for users to list the databases/schemas.
+ * If a databasePattern is supplied then the databases that only matches the
+ * pattern would be listed.
+ * The syntax of using this command in SQL is:
+ * {{{
+ *   SHOW (DATABASES|SCHEMAS) [LIKE 'identifier_with_wildcards'];
+ * }}}
+ */
+case class ShowDatabasesCommand(databasePattern: Option[String]) extends RunnableCommand {
+
+  // The result of SHOW DATABASES has one column called 'result'
+  override val output: Seq[Attribute] = {
+    AttributeReference("result", StringType, nullable = false)() :: Nil
+  }
+
+  override def run(sqlContext: SQLContext): Seq[Row] = {
+    val catalog = sqlContext.sessionState.catalog
+    val databases =
+      databasePattern.map(catalog.listDatabases(_)).getOrElse(catalog.listDatabases())
+    databases.map { d => Row(d) }
   }
 }
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
index cebf9c856d..458f36e832 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala
@@ -762,4 +762,15 @@ class DDLCommandSuite extends PlanTest {
     comparePlans(parsed2, expected2)
   }
 
+  test("show databases") {
+    val sql1 = "SHOW DATABASES"
+    val sql2 = "SHOW DATABASES LIKE 'defau*'"
+    val parsed1 = parser.parsePlan(sql1)
+    val expected1 = ShowDatabasesCommand(None)
+    val parsed2 = parser.parsePlan(sql2)
+    val expected2 = ShowDatabasesCommand(Some("defau*"))
+    comparePlans(parsed1, expected1)
+    comparePlans(parsed2, expected2)
+  }
+
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index f148f2d4ea..885a04af59 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -45,6 +45,7 @@ class DDLSuite extends QueryTest with SharedSQLContext {
       dbNames.foreach { name =>
         sqlContext.sql(s"DROP DATABASE IF EXISTS $name CASCADE")
       }
+      sqlContext.sessionState.catalog.setCurrentDatabase("default")
     }
   }
 
@@ -159,4 +160,75 @@ class DDLSuite extends QueryTest with SharedSQLContext {
   }
 
   // TODO: ADD a testcase for Drop Database in Restric when we can create tables in SQLContext
+
+  test("show tables") {
+    withTempTable("show1a", "show2b") {
+      sql(
+        """
+          |CREATE TEMPORARY TABLE show1a
+          |USING org.apache.spark.sql.sources.DDLScanSource
+          |OPTIONS (
+          |  From '1',
+          |  To '10',
+          |  Table 'test1'
+          |
+          |)
+        """.stripMargin)
+      sql(
+        """
+          |CREATE TEMPORARY TABLE show2b
+          |USING org.apache.spark.sql.sources.DDLScanSource
+          |OPTIONS (
+          |  From '1',
+          |  To '10',
+          |  Table 'test1'
+          |)
+        """.stripMargin)
+      checkAnswer(
+        sql("SHOW TABLES IN default 'show1*'"),
+        Row("show1a", true) :: Nil)
+
+      checkAnswer(
+        sql("SHOW TABLES IN default 'show1*|show2*'"),
+        Row("show1a", true) ::
+          Row("show2b", true) :: Nil)
+
+      checkAnswer(
+        sql("SHOW TABLES 'show1*|show2*'"),
+        Row("show1a", true) ::
+          Row("show2b", true) :: Nil)
+
+      assert(
+        sql("SHOW TABLES").count() >= 2)
+      assert(
+        sql("SHOW TABLES IN default").count() >= 2)
+    }
+  }
+
+  test("show databases") {
+    withDatabase("showdb1A", "showdb2B") {
+      sql("CREATE DATABASE showdb1A")
+      sql("CREATE DATABASE showdb2B")
+
+      assert(
+        sql("SHOW DATABASES").count() >= 2)
+
+      checkAnswer(
+        sql("SHOW DATABASES LIKE '*db1A'"),
+        Row("showdb1A") :: Nil)
+
+      checkAnswer(
+        sql("SHOW DATABASES LIKE 'showdb1A'"),
+        Row("showdb1A") :: Nil)
+
+      checkAnswer(
+        sql("SHOW DATABASES LIKE '*db1A|*db2B'"),
+        Row("showdb1A") ::
+          Row("showdb2B") :: Nil)
+
+      checkAnswer(
+        sql("SHOW DATABASES LIKE 'non-existentdb'"),
+        Nil)
+    }
+  }
 }
diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala
index 7ad7f92bd2..e93b0c145f 100644
--- a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala
+++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/CliSuite.scala
@@ -177,7 +177,7 @@ class CliSuite extends SparkFunSuite with BeforeAndAfterAll with Logging {
   }
 
   test("Single command with -e") {
-    runCliWithin(2.minute, Seq("-e", "SHOW DATABASES;"))("" -> "OK")
+    runCliWithin(2.minute, Seq("-e", "SHOW DATABASES;"))("" -> "")
   }
 
   test("Single command with --database") {
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
index 6199253d34..c203518fdd 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
@@ -1811,4 +1811,26 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       }
     }
   }
+
+  test("show tables") {
+    withTable("show1a", "show2b") {
+      sql("CREATE TABLE show1a(c1 int)")
+      sql("CREATE TABLE show2b(c2 int)")
+      checkAnswer(
+        sql("SHOW TABLES IN default 'show1*'"),
+        Row("show1a", false) :: Nil)
+      checkAnswer(
+        sql("SHOW TABLES IN default 'show1*|show2*'"),
+        Row("show1a", false) ::
+          Row("show2b", false) :: Nil)
+      checkAnswer(
+        sql("SHOW TABLES 'show1*|show2*'"),
+        Row("show1a", false) ::
+          Row("show2b", false) :: Nil)
+      assert(
+        sql("SHOW TABLES").count() >= 2)
+      assert(
+        sql("SHOW TABLES IN default").count() >= 2)
+    }
+  }
 }