[SPARK-53840][SQL] Add AS JSON output support for SHOW TABLES and SHOW TABLE EXTENDED

common/utils/src/main/resources/error/error-conditions.json

@@ -8295,6 +8295,11 @@
           "<variableName> is a VARIABLE and cannot be updated using the SET statement. Use SET VARIABLE <variableName> = ... instead."
         ]
       },
+      "SHOW_TABLE_EXTENDED_JSON_WITH_PARTITION" : {
+        "message" : [
+          "SHOW TABLE EXTENDED with PARTITION does not support AS JSON output."
+        ]
+      },
       "SQL_CURSOR" : {
         "message" : [
           "SQL cursor operations (DECLARE CURSOR, OPEN, FETCH, CLOSE) are not supported."

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

@@ -364,9 +364,9 @@ statement
     | EXPLAIN (LOGICAL | FORMATTED | EXTENDED | CODEGEN | COST)?
         (statement|setResetStatement)                                  #explain
     | SHOW TABLES ((FROM | IN) identifierReference)?
-        (LIKE? pattern=stringLit)?                                        #showTables
+        (LIKE? pattern=stringLit)? (AS JSON)?                            #showTables
     | SHOW TABLE EXTENDED ((FROM | IN) ns=identifierReference)?
-        LIKE pattern=stringLit partitionSpec?                             #showTableExtended
+        LIKE pattern=stringLit partitionSpec? (AS JSON)?                  #showTableExtended
     | SHOW TBLPROPERTIES table=identifierReference
         (LEFT_PAREN key=propertyKeyOrStringLit RIGHT_PAREN)?           #showTblProperties
     | SHOW COLUMNS (FROM | IN) table=identifierReference

sql/api/src/main/scala/org/apache/spark/sql/errors/CompilationErrors.scala

@@ -53,6 +53,12 @@ private[sql] trait CompilationErrors extends DataTypeErrorsBase {
       messageParameters = Map.empty)
   }
 
+  def showTableExtendedJsonWithPartitionError(): AnalysisException = {
+    new AnalysisException(
+      errorClass = "UNSUPPORTED_FEATURE.SHOW_TABLE_EXTENDED_JSON_WITH_PARTITION",
+      messageParameters = Map.empty)
+  }
+
   def cannotFindDescriptorFileError(filePath: String, cause: Throwable): AnalysisException = {
     new AnalysisException(
       errorClass = "PROTOBUF_DESCRIPTOR_FILE_NOT_FOUND",

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -1496,6 +1496,38 @@ class SparkSqlAstBuilder extends AstBuilder {
       Option(ctx.pattern).map(x => string(visitStringLit(x))))
   }
 
+  /**
+   * Create a [[ShowTablesJsonCommand]] or [[ShowTables]] command.
+   */
+  override def visitShowTables(ctx: ShowTablesContext): LogicalPlan = withOrigin(ctx) {
+    if (ctx.JSON == null) return super.visitShowTables(ctx)
+    val ns = if (ctx.identifierReference() != null) {
+      withIdentClause(ctx.identifierReference, UnresolvedNamespace(_))
+    } else {
+      CurrentNamespace
+    }
+    val pattern = Option(ctx.pattern).map(x => string(visitStringLit(x)))
+    ShowTablesJsonCommand(ns, pattern, isExtended = false)
+  }
+
+  /**
+   * Create a [[ShowTablesJsonCommand]], [[ShowTablesExtended]], or [[ShowTablePartition]] command.
+   */
+  override def visitShowTableExtended(
+      ctx: ShowTableExtendedContext): LogicalPlan = withOrigin(ctx) {
+    val asJson = ctx.JSON != null
+    if (asJson && ctx.partitionSpec != null) {
+      throw QueryCompilationErrors.showTableExtendedJsonWithPartitionError()
+    }
+    if (!asJson || ctx.partitionSpec != null) return super.visitShowTableExtended(ctx)
+    val ns = if (ctx.identifierReference() != null) {
+      withIdentClause(ctx.identifierReference, UnresolvedNamespace(_))
+    } else {
+      CurrentNamespace
+    }
+    ShowTablesJsonCommand(ns, Some(string(visitStringLit(ctx.pattern))), isExtended = true)
+  }
+
   override def visitDescribeProcedure(
       ctx: DescribeProcedureContext): LogicalPlan = withOrigin(ctx) {
     withIdentClause(ctx.identifierReference(), procIdentifier =>

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

@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.command
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.json4s._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.catalyst.analysis.ResolvedNamespace
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util.StringUtils
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, TableCatalog, TableViewCatalog}
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+import org.apache.spark.sql.types.{MetadataBuilder, StringType}
+
+/**
+ * The command for `SHOW TABLES AS JSON` and `SHOW TABLE EXTENDED AS JSON`.
+ */
+case class ShowTablesJsonCommand(
+    child: LogicalPlan,
+    pattern: Option[String],
+    isExtended: Boolean,
+    override val output: Seq[Attribute] = Seq(
+      AttributeReference(
+        "json_metadata",
+        StringType,
+        nullable = false,
+        new MetadataBuilder()
+          .putString("comment", "JSON metadata of the tables")
+          .build())()
+    )) extends UnaryRunnableCommand {
+
+  override def run(sparkSession: SparkSession): Seq[Row] = {
+    val jsonOutput = child match {
+      case ResolvedNamespace(catalog, ns, _) =>
+        if (CatalogV2Util.isSessionCatalog(catalog)) {
+          runForSessionCatalog(sparkSession, ns)
+        } else {
+          runForV2Catalog(sparkSession, catalog.asTableCatalog, ns)
+        }
+      case other =>
+        throw SparkException.internalError(
+          s"Unexpected child in ShowTablesJsonCommand: ${other.getClass.getSimpleName}")
+    }
+    Seq(Row(compact(render(jsonOutput))))
+  }
+
+  private def runForSessionCatalog(
+      sparkSession: SparkSession,
+      ns: Seq[String]): JObject = {
+    val sessionCatalog = sparkSession.sessionState.catalog
+    val db = ns.headOption.getOrElse(sessionCatalog.getCurrentDatabase)
+    val tables = pattern
+      .map(p => sessionCatalog.listTables(db, p))
+      .getOrElse(sessionCatalog.listTables(db))
+
+    val jsonTables = tables.map { tableIdent =>
+      val isTemp = sessionCatalog.isTempView(tableIdent)
+      val namespace = tableIdent.database.toList
+
+      if (isExtended) {
+        val tableType = if (isTemp) {
+          "VIEW"
+        } else {
+          sessionCatalog.getTempViewOrPermanentTableMetadata(tableIdent).tableType.name
+        }
+        JObject(
+          "name" -> JString(tableIdent.table),
+          "catalog" -> JString(
+            sparkSession.sessionState.catalogManager.v2SessionCatalog.name()),
+          "namespace" -> JArray(namespace.map(JString(_))),
+          "type" -> JString(tableType),
+          "isTemporary" -> JBool(isTemp)
+        )
+      } else {
+        JObject(
+          "name" -> JString(tableIdent.table),
+          "namespace" -> JArray(namespace.map(JString(_))),
+          "isTemporary" -> JBool(isTemp)
+        )
+      }
+    }.toList
+
+    JObject("tables" -> JArray(jsonTables))
+  }
+
+  private def runForV2Catalog(
+      sparkSession: SparkSession,
+      catalog: TableCatalog,
+      ns: Seq[String]): JObject = {
+    val identifiers = if (!isExtended) {
+      catalog match {
+        case mc: TableViewCatalog =>
+          mc.listTableAndViewSummaries(ns.toArray).map(_.identifier())
+        case _ => catalog.listTables(ns.toArray)
+      }
+    } else {
+      catalog.listTables(ns.toArray)
+    }
+
+    val pat = pattern.getOrElse("*")
+    val filteredIdents = identifiers.filter { ident =>
+      StringUtils.filterPattern(Seq(ident.name()), pat).nonEmpty
+    }
+
+    val jsonRows = new ArrayBuffer[JObject]()
+    filteredIdents.foreach { ident =>
+      val nsArray = JArray(ident.namespace().map(JString(_)).toList)
+      val entry = if (isExtended) {
+        JObject(
+          "name" -> JString(ident.name()),
+          "catalog" -> JString(catalog.name()),
+          "namespace" -> nsArray,
+          "type" -> JString("TABLE"),
+          "isTemporary" -> JBool(false)
+        )
+      } else {
+        JObject(
+          "name" -> JString(ident.name()),
+          "namespace" -> nsArray,
+          "isTemporary" -> JBool(false)
+        )
+      }
+      jsonRows += entry
+    }
+
+    JObject("tables" -> JArray(jsonRows.toList))
+  }
+
+  override protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan =
+    copy(child = newChild)
+}

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

@@ -17,9 +17,11 @@
 
 package org.apache.spark.sql.execution.command
 
+import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, CurrentNamespace, UnresolvedNamespace, UnresolvedPartitionSpec, UnresolvedTable}
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser.parsePlan
 import org.apache.spark.sql.catalyst.plans.logical.{ShowTablePartition, ShowTables, ShowTablesExtended}
+import org.apache.spark.sql.execution.command.ShowTablesJsonCommand
 import org.apache.spark.sql.test.SharedSparkSession
 
 class ShowTablesParserSuite extends AnalysisTest with SharedSparkSession {
@@ -49,6 +51,19 @@ class ShowTablesParserSuite extends AnalysisTest with SharedSparkSession {
       ShowTables(UnresolvedNamespace(Seq("ns1")), Some("*test*")))
   }
 
+  test("show tables as json") {
+    val parse = spark.sessionState.sqlParser.parsePlan _
+    comparePlans(
+      parse("SHOW TABLES AS JSON"),
+      ShowTablesJsonCommand(CurrentNamespace, None, isExtended = false))
+    comparePlans(
+      parse("SHOW TABLES IN ns1 AS JSON"),
+      ShowTablesJsonCommand(UnresolvedNamespace(Seq("ns1")), None, isExtended = false))
+    comparePlans(
+      parse("SHOW TABLES IN ns1 LIKE '*test*' AS JSON"),
+      ShowTablesJsonCommand(UnresolvedNamespace(Seq("ns1")), Some("*test*"), isExtended = false))
+  }
+
   test("show table extended") {
     comparePlans(
       parsePlan("SHOW TABLE EXTENDED LIKE '*test*'"),
@@ -80,4 +95,26 @@ class ShowTablesParserSuite extends AnalysisTest with SharedSparkSession {
           "SHOW TABLE EXTENDED ... PARTITION ..."),
         UnresolvedPartitionSpec(Map("ds" -> "2008-04-09"))))
   }
+
+  test("show table extended as json") {
+    val parse = spark.sessionState.sqlParser.parsePlan _
+    comparePlans(
+      parse("SHOW TABLE EXTENDED LIKE '*test*' AS JSON"),
+      ShowTablesJsonCommand(CurrentNamespace, Some("*test*"), isExtended = true))
+    comparePlans(
+      parse(s"SHOW TABLE EXTENDED IN $catalog.ns1.ns2 LIKE '*test*' AS JSON"),
+      ShowTablesJsonCommand(
+        UnresolvedNamespace(Seq(catalog, "ns1", "ns2")), Some("*test*"), isExtended = true))
+  }
+
+  test("show table extended as json with partition should fail") {
+    checkError(
+      exception = intercept[AnalysisException] {
+        spark.sessionState.sqlParser.parsePlan(
+          "SHOW TABLE EXTENDED LIKE '*test*' PARTITION(ds='2008-04-09') AS JSON")
+      },
+      condition = "UNSUPPORTED_FEATURE.SHOW_TABLE_EXTENDED_JSON_WITH_PARTITION",
+      parameters = Map.empty
+    )
+  }
 }