[SPARK-33393][SQL] Support SHOW TABLE EXTENDED in v2

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -206,10 +206,6 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog with QueryErrorsB
           case _ =>
         }
 
-      // `ShowTableExtended` should have been converted to the v1 command if the table is v1.
-      case _: ShowTableExtended =>
-        throw QueryCompilationErrors.commandUnsupportedInV2TableError("SHOW TABLE EXTENDED")
-
       case operator: LogicalPlan =>
         operator transformExpressionsDown {
           // Check argument data types of higher-order functions downwards first.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolvePartitionSpec.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.analysis
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2PartitionCommand}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, ShowTableExtended, V2PartitionCommand}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.COMMAND
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
@@ -50,6 +50,10 @@ object ResolvePartitionSpec extends Rule[LogicalPlan] {
           }
         case _ => command
       }
+    case s @ ShowTableExtended(_, _, partitionSpec @ Some(UnresolvedPartitionSpec(_, _)), _) =>
+      val extractPartitionSpec = new ExtractPartitionSpec(
+        partitionSpec.get.asInstanceOf[UnresolvedPartitionSpec])
+      s.copy(partitionSpec = Some(extractPartitionSpec))
   }
 
   private def resolvePartitionSpec(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/v2ResolutionPlans.scala

@@ -93,6 +93,15 @@ case class UnresolvedPartitionSpec(
   override lazy val resolved = false
 }
 
+case class ExtractPartitionSpec(
+    spec: Map[String, String],
+    location: Option[String] = None) extends PartitionSpec {
+
+  def this(u: UnresolvedPartitionSpec) = this(u.spec, u.location)
+
+  override lazy val resolved = true
+}
+
 sealed trait FieldName extends LeafExpression with Unevaluable {
   def name: Seq[String]
   override def dataType: DataType = throw new IllegalStateException(

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -2016,12 +2016,6 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
         "config" -> LEGACY_CTE_PRECEDENCE_POLICY.key))
   }
 
-  def commandUnsupportedInV2TableError(name: String): Throwable = {
-    new AnalysisException(
-      errorClass = "_LEGACY_ERROR_TEMP_1200",
-      messageParameters = Map("name" -> name))
-  }
-
   def cannotResolveColumnNameAmongAttributesError(
       colName: String, fieldNames: String): Throwable = {
     new AnalysisException(

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -32,9 +32,9 @@ import org.codehaus.commons.compiler.{CompileException, InternalCompilerExceptio
 import org.apache.spark._
 import org.apache.spark.launcher.SparkLauncher
 import org.apache.spark.memory.SparkOutOfMemoryError
-import org.apache.spark.sql.catalyst.{TableIdentifier, WalkedTypePath}
+import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier, WalkedTypePath}
 import org.apache.spark.sql.catalyst.ScalaReflection.Schema
-import org.apache.spark.sql.catalyst.analysis.UnresolvedGenerator
+import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException, UnresolvedGenerator}
 import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable}
 import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
 import org.apache.spark.sql.catalyst.parser.ParseException
@@ -2790,4 +2790,17 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
         "location" -> toSQLValue(location.toString, StringType),
         "identifier" -> toSQLId(tableId.nameParts)))
   }
+
+  def showTableExtendedMultiPartitionUnsupportedError(tableName: String): Throwable = {
+    new UnsupportedOperationException(
+      s"The table $tableName does not support show table extended of multiple partition.")
+  }
+
+  def notExistPartitionError(
+      tableName: String,
+      partitionIdent: InternalRow,
+      partitionSchema: StructType): Throwable = {
+    new NoSuchPartitionException(tableName, partitionIdent, partitionSchema)
+  }
+
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Implicits.scala

@@ -65,6 +65,11 @@ object DataSourceV2Implicits {
       }
     }
 
+    def isPartitionable: Boolean = table match {
+      case _: SupportsPartitionManagement => true
+      case _ => false
+    }
+
     def asPartitionable: SupportsPartitionManagement = {
       table match {
         case support: SupportsPartitionManagement =>

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -256,15 +256,20 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
     case ShowTableExtended(
         DatabaseInSessionCatalog(db),
         pattern,
-        partitionSpec @ (None | Some(UnresolvedPartitionSpec(_, _))),
+        partitionSpec @ (None | Some(UnresolvedPartitionSpec(_, _)) |
+                         Some(ExtractPartitionSpec(_, _))),
         output) =>
       val newOutput = if (conf.getConf(SQLConf.LEGACY_KEEP_COMMAND_OUTPUT_SCHEMA)) {
         assert(output.length == 4)
         output.head.withName("database") +: output.tail
       } else {
         output
       }
-      val tablePartitionSpec = partitionSpec.map(_.asInstanceOf[UnresolvedPartitionSpec].spec)
+      val tablePartitionSpec = partitionSpec match {
+        case Some(UnresolvedPartitionSpec(spec, _)) => Some(spec)
+        case Some(ExtractPartitionSpec(spec, _)) => Some(spec)
+        case _ => None
+      }
       ShowTablesCommand(Some(db), Some(pattern), newOutput, true, tablePartitionSpec)
 
     // ANALYZE TABLE works on permanent views if the views are cached.

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

@@ -24,7 +24,7 @@ import org.apache.commons.lang3.StringUtils
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{SparkSession, Strategy}
-import org.apache.spark.sql.catalyst.analysis.{ResolvedIdentifier, ResolvedNamespace, ResolvedPartitionSpec, ResolvedTable}
+import org.apache.spark.sql.catalyst.analysis.{ExtractPartitionSpec, ResolvedIdentifier, ResolvedNamespace, ResolvedPartitionSpec, ResolvedTable}
 import org.apache.spark.sql.catalyst.catalog.CatalogUtils
 import org.apache.spark.sql.catalyst.expressions
 import org.apache.spark.sql.catalyst.expressions.{And, Attribute, DynamicPruning, Expression, NamedExpression, Not, Or, PredicateHelper, SubqueryExpression}
@@ -395,6 +395,20 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
     case ShowTables(ResolvedNamespace(catalog, ns), pattern, output) =>
       ShowTablesExec(output, catalog.asTableCatalog, ns, pattern) :: Nil
 
+    case ShowTableExtended(
+        ResolvedNamespace(catalog, ns),
+        pattern,
+        partitionSpec @ (None | Some(ExtractPartitionSpec(_, _))),
+        output) =>
+      val tablePartitionSpec = partitionSpec.map(_.asInstanceOf[ExtractPartitionSpec].spec)
+      ShowTablesExec(
+        output,
+        catalog.asTableCatalog,
+        ns,
+        Some(pattern),
+        true,
+        tablePartitionSpec) :: Nil
+
     case SetCatalogAndNamespace(ResolvedNamespace(catalog, ns)) =>
       val catalogManager = session.sessionState.catalogManager
       val namespace = if (ns.nonEmpty) Some(ns) else None

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowTablesExec.scala

@@ -17,14 +17,25 @@
 
 package org.apache.spark.sql.execution.datasources.v2
 
+import scala.collection.JavaConverters._
+import scala.collection.convert.ImplicitConversions.`map AsScala`
+import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.util.StringUtils
-import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
-import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.NamespaceHelper
+import org.apache.spark.sql.catalyst.analysis.ResolvePartitionSpec.convertToPartIdent
+import org.apache.spark.sql.catalyst.catalog.CatalogTableType
+import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
+import org.apache.spark.sql.catalyst.catalog.ExternalCatalogUtils.escapePathName
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Literal}
+import org.apache.spark.sql.catalyst.util.{quoteIdentifier, StringUtils}
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier, SupportsPartitionManagement, Table, TableCatalog}
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.execution.LeafExecNode
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits.TableHelper
+import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.util.PartitioningUtils.{normalizePartitionSpec, requireExactMatchedPartitionSpec}
 
 /**
  * Physical plan node for showing tables.
@@ -33,14 +44,37 @@ case class ShowTablesExec(
     output: Seq[Attribute],
     catalog: TableCatalog,
     namespace: Seq[String],
-    pattern: Option[String]) extends V2CommandExec with LeafExecNode {
+    pattern: Option[String],
+    isExtended: Boolean = false,
+    partitionSpec: Option[TablePartitionSpec] = None) extends V2CommandExec with LeafExecNode {
   override protected def run(): Seq[InternalRow] = {
     val rows = new ArrayBuffer[InternalRow]()
 
-    val tables = catalog.listTables(namespace.toArray)
-    tables.map { table =>
-      if (pattern.map(StringUtils.filterPattern(Seq(table.name()), _).nonEmpty).getOrElse(true)) {
-        rows += toCatalystRow(table.namespace().quoted, table.name(), isTempView(table))
+    val identifiers = catalog.listTables(namespace.toArray)
+    identifiers.map { identifier =>
+      if (pattern.map(StringUtils.filterPattern(
+        Seq(identifier.name()), _).nonEmpty).getOrElse(true)) {
+        if (!isExtended) {
+          rows += toCatalystRow(identifier.namespace().quoted, identifier.name(),
+            isTempView(identifier))
+        } else {
+          val table = catalog.loadTable(identifier)
+          if (!partitionSpec.isEmpty && table.isPartitionable) {
+            // Show the information of partitions.
+            rows += toCatalystRow(
+              identifier.namespace().quoted,
+              identifier.name(),
+              isTempView(identifier),
+              s"${extendedPartition(identifier, table.asPartitionable, partitionSpec)}")
+          } else {
+            // Show the information of tables.
+            rows += toCatalystRow(
+              identifier.namespace().quoted,
+              identifier.name(),
+              isTempView(identifier),
+              s"${extendedTable(identifier, table)}")
+          }
+        }
       }
     }
 
@@ -53,4 +87,113 @@ case class ShowTablesExec(
       case _ => false
     }
   }
+
+  private def extendedTable(identifier: Identifier, table: Table): String = {
+    val results = new mutable.LinkedHashMap[String, String]()
+
+    if (!identifier.namespace().isEmpty) {
+      results.put("Namespace", identifier.namespace().quoted)
+    }
+    results.put("Table", identifier.name())
+    val tableType = if (table.properties().containsKey(TableCatalog.PROP_EXTERNAL)) {
+      CatalogTableType.EXTERNAL
+    } else {
+      CatalogTableType.MANAGED
+    }
+    results.put("Type", tableType.name)
+
+    CatalogV2Util.TABLE_RESERVED_PROPERTIES
+      .filterNot(_ == TableCatalog.PROP_EXTERNAL)
+      .foreach(propKey => {
+        if (table.properties.containsKey(propKey)) {
+          results.put(propKey.capitalize, table.properties.get(propKey))
+        }
+      })
+
+    val properties =
+      conf.redactOptions(table.properties.asScala.toMap).toList
+        .filter(kv => !CatalogV2Util.TABLE_RESERVED_PROPERTIES.contains(kv._1))
+        .sortBy(_._1).map {
+        case (key, value) => key + "=" + value
+      }.mkString("[", ",", "]")
+    if (table.properties().isEmpty) {
+      results.put("Table Properties", properties.mkString("[", ", ", "]"))
+    }
+
+    // Partition Provider & Partition Columns
+    // TODO check
+    if (table.isPartitionable && !table.asPartitionable.partitionSchema().isEmpty) {
+      results.put("Partition Provider", "Catalog")
+      results.put("Partition Columns", table.asPartitionable.partitionSchema().map(
+        field => quoteIdentifier(field.name)).mkString(", "))
+    }
+
+    if (table.schema().nonEmpty) results.put("Schema", table.schema().treeString)
+
+    results.map { case ((key, value)) =>
+      if (value.isEmpty) key else s"$key: $value"
+    }.mkString("", "\n", "")
+  }
+
+  private def extendedPartition(
+      identifier: Identifier,
+      partitionTable: SupportsPartitionManagement,
+      partitionSpec: Option[TablePartitionSpec]): String = {
+    val results = new mutable.LinkedHashMap[String, String]()
+
+    // "Partition Values"
+    val partitionSchema = partitionTable.partitionSchema()
+    val normalizedSpec = normalizePartitionSpec(
+      partitionSpec.get,
+      partitionSchema,
+      partitionTable.name(),
+      conf.resolver)
+    requireExactMatchedPartitionSpec(identifier.toString,
+      normalizedSpec, partitionSchema.fieldNames)
+
+    val partitionNames = normalizedSpec.keySet
+    val (names, ident) = (partitionSchema.map(_.name),
+      convertToPartIdent(normalizedSpec, partitionSchema))
+    val partitionIdentifiers = partitionTable.listPartitionIdentifiers(names.toArray, ident)
+    partitionIdentifiers.length match {
+      case 0 =>
+        throw QueryExecutionErrors.notExistPartitionError(
+          identifier.toString, ident, partitionSchema)
+      case len if (len > 1) =>
+        throw QueryExecutionErrors.showTableExtendedMultiPartitionUnsupportedError(
+          identifier.toString)
+      case _ => // do nothing
+    }
+    val partitionIdentifier = partitionIdentifiers.head
+    val len = partitionSchema.length
+    val partitions = new Array[String](len)
+    val timeZoneId = conf.sessionLocalTimeZone
+    var i = 0
+    while (i < len) {
+      val dataType = partitionSchema(i).dataType
+      val partValueUTF8String =
+        Cast(Literal(partitionIdentifier.get(i, dataType), dataType),
+          StringType, Some(timeZoneId)).eval()
+      val partValueStr = if (partValueUTF8String == null) "null" else partValueUTF8String.toString
+      partitions(i) = escapePathName(partitionSchema(i).name) + "=" + escapePathName(partValueStr)
+      i += 1
+    }
+    val partitionValues = partitions.mkString("[", ", ", "]")
+    results.put("Partition Values", s"${partitionValues}")
+
+    // "Partition Parameters"
+    val metadata = partitionTable.loadPartitionMetadata(ident)
+    if (!metadata.isEmpty) {
+      val metadataValues = metadata.map { case ((key, value)) =>
+        if (value.isEmpty) key else s"$key: $value"
+      }.mkString("{", ", ", "}")
+      results.put("Partition Parameters", metadataValues)
+    }
+
+    // TODO "Created Time", "Last Access", "Partition Statistics"
+
+    results.map { case ((key, value)) =>
+      if (value.isEmpty) key else s"$key: $value"
+    }.mkString("", "\n", "\n")
+  }
 }