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

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

@@ -109,7 +109,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)?
@@ -622,7 +623,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
@@ -840,6 +841,7 @@ OUTPUTFORMAT: 'OUTPUTFORMAT';
 INPUTDRIVER: 'INPUTDRIVER';
 OUTPUTDRIVER: 'OUTPUTDRIVER';
 DATABASE: 'DATABASE' | 'SCHEMA';
+DATABASES: 'DATABASES' | 'SCHEMAS';
 DFS: 'DFS';
 TRUNCATE: 'TRUNCATE';
 METADATA: 'METADATA';

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))
   }
 
   /**

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))
   }
 
   /**

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

@@ -322,30 +322,54 @@ 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] = {
     // Since we need to return a Seq of rows, we will call getTables directly
     // 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) }
   }
 }
 

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

@@ -663,4 +663,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)
+  }
+
 }

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)
+    }
+  }
 }

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") {

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)
+    }
+  }
 }