[SPARK-18961][SQL] Support `SHOW TABLE EXTENDED ... PARTITION` statement

## What changes were proposed in this pull request?

We should support the statement `SHOW TABLE EXTENDED LIKE 'table_identifier' PARTITION(partition_spec)`, just like that HIVE does.
When partition is specified, the `SHOW TABLE EXTENDED` command should output the information of the partitions instead of the tables.
Note that in this statement, we require exact matched partition spec. For example:
```
CREATE TABLE show_t1(a String, b Int) PARTITIONED BY (c String, d String);
ALTER TABLE show_t1 ADD PARTITION (c='Us', d=1) PARTITION (c='Us', d=22);

-- Output the extended information of Partition(c='Us', d=1)
SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Us', d=1);
-- Throw an AnalysisException
SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Us');
```

## How was this patch tested?
Add new test sqls in file `show-tables.sql`.
Add new test case in `DDLSuite`.

Author: jiangxingbo <jiangxb1987@gmail.com>

Closes #16373 from jiangxb1987/show-partition-extended.

7 files changed, 163 insertions, 83 deletions

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
index 9a3656ddc7..8e8210e334 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
@@ -127,8 +127,8 @@ class QueryExecution(val sparkSession: SparkSession, val logical: LogicalPlan) {
             .map(s => String.format(s"%-20s", s))
             .mkString("\t")
       }
-    // SHOW TABLES in Hive only output table names, while ours outputs database, table name, isTemp.
-    case command: ExecutedCommandExec if command.cmd.isInstanceOf[ShowTablesCommand] =>
+    // SHOW TABLES in Hive only output table names, while ours output database, table name, isTemp.
+    case command @ ExecutedCommandExec(s: ShowTablesCommand) if !s.isExtended =>
       command.executeCollect().map(_.getString(1))
     case other =>
       val result: Seq[Seq[Any]] = other.executeCollectPublic().map(_.toSeq).toSeq
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 00d1d6d270..abea7a3bcf 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
@@ -134,7 +134,8 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
     ShowTablesCommand(
       Option(ctx.db).map(_.getText),
       Option(ctx.pattern).map(string),
-      isExtended = false)
+      isExtended = false,
+      partitionSpec = None)
   }
 
   /**
@@ -146,14 +147,12 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
    * }}}
    */
   override def visitShowTable(ctx: ShowTableContext): LogicalPlan = withOrigin(ctx) {
-    if (ctx.partitionSpec != null) {
-      operationNotAllowed("SHOW TABLE EXTENDED ... PARTITION", ctx)
-    }
-
+    val partitionSpec = Option(ctx.partitionSpec).map(visitNonOptionalPartitionSpec)
     ShowTablesCommand(
       Option(ctx.db).map(_.getText),
       Option(ctx.pattern).map(string),
-      isExtended = true)
+      isExtended = true,
+      partitionSpec = partitionSpec)
   }
 
   /**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 86394ff23e..beb3dcafd6 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -616,13 +616,15 @@ case class DescribeTableCommand(
  * The syntax of using this command in SQL is:
  * {{{
  *   SHOW TABLES [(IN|FROM) database_name] [[LIKE] 'identifier_with_wildcards'];
- *   SHOW TABLE EXTENDED [(IN|FROM) database_name] LIKE 'identifier_with_wildcards';
+ *   SHOW TABLE EXTENDED [(IN|FROM) database_name] LIKE 'identifier_with_wildcards'
+ *   [PARTITION(partition_spec)];
  * }}}
  */
 case class ShowTablesCommand(
     databaseName: Option[String],
     tableIdentifierPattern: Option[String],
-    isExtended: Boolean = false) extends RunnableCommand {
+    isExtended: Boolean = false,
+    partitionSpec: Option[TablePartitionSpec] = None) extends RunnableCommand {
 
   // The result of SHOW TABLES/SHOW TABLE has three basic columns: database, tableName and
   // isTemporary. If `isExtended` is true, append column `information` to the output columns.
@@ -642,18 +644,34 @@ case class ShowTablesCommand(
     // instead of calling tables in sparkSession.
     val catalog = sparkSession.sessionState.catalog
     val db = databaseName.getOrElse(catalog.getCurrentDatabase)
-    val tables =
-      tableIdentifierPattern.map(catalog.listTables(db, _)).getOrElse(catalog.listTables(db))
-    tables.map { tableIdent =>
-      val database = tableIdent.database.getOrElse("")
-      val tableName = tableIdent.table
-      val isTemp = catalog.isTemporaryTable(tableIdent)
-      if (isExtended) {
-        val information = catalog.getTempViewOrPermanentTableMetadata(tableIdent).toString
-        Row(database, tableName, isTemp, s"${information}\n")
-      } else {
-        Row(database, tableName, isTemp)
+    if (partitionSpec.isEmpty) {
+      // Show the information of tables.
+      val tables =
+        tableIdentifierPattern.map(catalog.listTables(db, _)).getOrElse(catalog.listTables(db))
+      tables.map { tableIdent =>
+        val database = tableIdent.database.getOrElse("")
+        val tableName = tableIdent.table
+        val isTemp = catalog.isTemporaryTable(tableIdent)
+        if (isExtended) {
+          val information = catalog.getTempViewOrPermanentTableMetadata(tableIdent).toString
+          Row(database, tableName, isTemp, s"$information\n")
+        } else {
+          Row(database, tableName, isTemp)
+        }
       }
+    } else {
+      // Show the information of partitions.
+      //
+      // Note: tableIdentifierPattern should be non-empty, otherwise a [[ParseException]]
+      // should have been thrown by the sql parser.
+      val tableIdent = TableIdentifier(tableIdentifierPattern.get, Some(db))
+      val table = catalog.getTableMetadata(tableIdent).identifier
+      val partition = catalog.getPartition(tableIdent, partitionSpec.get)
+      val database = table.database.getOrElse("")
+      val tableName = table.table
+      val isTemp = catalog.isTemporaryTable(table)
+      val information = partition.toString
+      Seq(Row(database, tableName, isTemp, s"$information\n"))
     }
   }
 }
diff --git a/sql/core/src/test/resources/sql-tests/inputs/show-tables.sql b/sql/core/src/test/resources/sql-tests/inputs/show-tables.sql
index 10c379dfa0..3c77c9977d 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/show-tables.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/show-tables.sql
@@ -17,10 +17,21 @@ SHOW TABLES LIKE 'show_t1*|show_t2*';
 SHOW TABLES IN showdb 'show_t*';
 
 -- SHOW TABLE EXTENDED
--- Ignore these because there exist timestamp results, e.g. `Created`.
--- SHOW TABLE EXTENDED LIKE 'show_t*';
+SHOW TABLE EXTENDED LIKE 'show_t*';
 SHOW TABLE EXTENDED;
+
+-- SHOW TABLE EXTENDED ... PARTITION
+SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Us', d=1);
+-- Throw a ParseException if table name is not specified.
+SHOW TABLE EXTENDED PARTITION(c='Us', d=1);
+-- Don't support regular expression for table name if a partition specification is present.
+SHOW TABLE EXTENDED LIKE 'show_t*' PARTITION(c='Us', d=1);
+-- Partition specification is not complete.
 SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Us');
+-- Partition specification is invalid.
+SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(a='Us', d=1);
+-- Partition specification doesn't exist.
+SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Ch', d=1);
 
 -- Clean Up
 DROP TABLE show_t1;
diff --git a/sql/core/src/test/resources/sql-tests/results/show-tables.sql.out b/sql/core/src/test/resources/sql-tests/results/show-tables.sql.out
index 3d287f43ac..6d62e60921 100644
--- a/sql/core/src/test/resources/sql-tests/results/show-tables.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/show-tables.sql.out
@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 20
+-- Number of queries: 26
 
 
 -- !query 0
@@ -114,76 +114,159 @@ show_t3
 
 
 -- !query 12
-SHOW TABLE EXTENDED
+SHOW TABLE EXTENDED LIKE 'show_t*'
 -- !query 12 schema
-struct<>
+struct<database:string,tableName:string,isTemporary:boolean,information:string>
 -- !query 12 output
-org.apache.spark.sql.catalyst.parser.ParseException
-
-mismatched input '<EOF>' expecting 'LIKE'(line 1, pos 19)
-
-== SQL ==
-SHOW TABLE EXTENDED
--------------------^^^
+show_t3	true	CatalogTable(
+	Table: `show_t3`
+	Created: 
+	Last Access: 
+	Type: VIEW
+	Schema: [StructField(e,IntegerType,true)]
+	Storage())
+
+showdb	show_t1	false	CatalogTable(
+	Table: `showdb`.`show_t1`
+	Created: 
+	Last Access: 
+	Type: MANAGED
+	Schema: [StructField(a,StringType,true), StructField(b,IntegerType,true), StructField(c,StringType,true), StructField(d,StringType,true)]
+	Provider: parquet
+	Partition Columns: [`c`, `d`]
+	Storage(Location: sql/core/spark-warehouse/showdb.db/show_t1)
+	Partition Provider: Catalog)
+
+showdb	show_t2	false	CatalogTable(
+	Table: `showdb`.`show_t2`
+	Created: 
+	Last Access: 
+	Type: MANAGED
+	Schema: [StructField(b,StringType,true), StructField(d,IntegerType,true)]
+	Provider: parquet
+	Storage(Location: sql/core/spark-warehouse/showdb.db/show_t2))
 
 
 -- !query 13
-SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Us')
+SHOW TABLE EXTENDED
 -- !query 13 schema
 struct<>
 -- !query 13 output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-Operation not allowed: SHOW TABLE EXTENDED ... PARTITION(line 1, pos 0)
+mismatched input '<EOF>' expecting 'LIKE'(line 1, pos 19)
 
 == SQL ==
-SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Us')
-^^^
+SHOW TABLE EXTENDED
+-------------------^^^
 
 
 -- !query 14
-DROP TABLE show_t1
+SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Us', d=1)
 -- !query 14 schema
-struct<>
+struct<database:string,tableName:string,isTemporary:boolean,information:string>
 -- !query 14 output
-
+showdb	show_t1	false	CatalogPartition(
+	Partition Values: [c=Us, d=1]
+	Storage(Location: sql/core/spark-warehouse/showdb.db/show_t1/c=Us/d=1)
+	Partition Parameters:{})
 
 
 -- !query 15
-DROP TABLE show_t2
+SHOW TABLE EXTENDED PARTITION(c='Us', d=1)
 -- !query 15 schema
 struct<>
 -- !query 15 output
+org.apache.spark.sql.catalyst.parser.ParseException
+
+mismatched input 'PARTITION' expecting 'LIKE'(line 1, pos 20)
 
+== SQL ==
+SHOW TABLE EXTENDED PARTITION(c='Us', d=1)
+--------------------^^^
 
 
 -- !query 16
-DROP VIEW  show_t3
+SHOW TABLE EXTENDED LIKE 'show_t*' PARTITION(c='Us', d=1)
 -- !query 16 schema
 struct<>
 -- !query 16 output
-
+org.apache.spark.sql.catalyst.analysis.NoSuchTableException
+Table or view 'show_t*' not found in database 'showdb';
 
 
 -- !query 17
-DROP VIEW  global_temp.show_t4
+SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Us')
 -- !query 17 schema
 struct<>
 -- !query 17 output
-
+org.apache.spark.sql.AnalysisException
+Partition spec is invalid. The spec (c) must match the partition spec (c, d) defined in table '`showdb`.`show_t1`';
 
 
 -- !query 18
-USE default
+SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(a='Us', d=1)
 -- !query 18 schema
 struct<>
 -- !query 18 output
-
+org.apache.spark.sql.AnalysisException
+Partition spec is invalid. The spec (a, d) must match the partition spec (c, d) defined in table '`showdb`.`show_t1`';
 
 
 -- !query 19
-DROP DATABASE showdb
+SHOW TABLE EXTENDED LIKE 'show_t1' PARTITION(c='Ch', d=1)
 -- !query 19 schema
 struct<>
 -- !query 19 output
+org.apache.spark.sql.catalyst.analysis.NoSuchPartitionException
+Partition not found in table 'show_t1' database 'showdb':
+c -> Ch
+d -> 1;
+
+
+-- !query 20
+DROP TABLE show_t1
+-- !query 20 schema
+struct<>
+-- !query 20 output
+
+
+
+-- !query 21
+DROP TABLE show_t2
+-- !query 21 schema
+struct<>
+-- !query 21 output
+
+
+
+-- !query 22
+DROP VIEW  show_t3
+-- !query 22 schema
+struct<>
+-- !query 22 output
+
+
+
+-- !query 23
+DROP VIEW  global_temp.show_t4
+-- !query 23 schema
+struct<>
+-- !query 23 output
+
+
+
+-- !query 24
+USE default
+-- !query 24 schema
+struct<>
+-- !query 24 output
+
+
+
+-- !query 25
+DROP DATABASE showdb
+-- !query 25 schema
+struct<>
+-- !query 25 output
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
index 68ababcd11..c285995514 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
@@ -222,7 +222,10 @@ class SQLQueryTestSuite extends QueryTest with SharedSQLContext {
       val df = session.sql(sql)
       val schema = df.schema
       // Get answer, but also get rid of the #1234 expression ids that show up in explain plans
-      val answer = df.queryExecution.hiveResultString().map(_.replaceAll("#\\d+", "#x"))
+      val answer = df.queryExecution.hiveResultString().map(_.replaceAll("#\\d+", "#x")
+        .replaceAll("Location: .*/sql/core/", "Location: sql/core/")
+        .replaceAll("Created: .*\n", "Created: \n")
+        .replaceAll("Last Access: .*\n", "Last Access: \n"))
 
       // If the output is not pre-sorted, sort it.
       if (isSorted(df.queryExecution.analyzed)) (schema, answer) else (schema, answer.sorted)
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 0666f446f3..6eed10ec51 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
@@ -977,40 +977,6 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
     testRenamePartitions(isDatasourceTable = false)
   }
 
-  test("show table extended") {
-    withTempView("show1a", "show2b") {
-      sql(
-        """
-         |CREATE TEMPORARY VIEW show1a
-         |USING org.apache.spark.sql.sources.DDLScanSource
-         |OPTIONS (
-         |  From '1',
-         |  To '10',
-         |  Table 'test1'
-         |
-         |)
-        """.stripMargin)
-      sql(
-        """
-         |CREATE TEMPORARY VIEW show2b
-         |USING org.apache.spark.sql.sources.DDLScanSource
-         |OPTIONS (
-         |  From '1',
-         |  To '10',
-         |  Table 'test1'
-         |)
-        """.stripMargin)
-      assert(
-        sql("SHOW TABLE EXTENDED LIKE 'show*'").count() >= 2)
-      assert(
-        sql("SHOW TABLE EXTENDED LIKE 'show*'").schema ==
-          StructType(StructField("database", StringType, false) ::
-            StructField("tableName", StringType, false) ::
-            StructField("isTemporary", BooleanType, false) ::
-            StructField("information", StringType, false) :: Nil))
-    }
-  }
-
   test("show databases") {
     sql("CREATE DATABASE showdb2B")
     sql("CREATE DATABASE showdb1A")