[SPARK-51615][SQL] Refactor ShowNamespaces to use RunnableCommand

### What changes were proposed in this pull request?
Refactor ShowNamespaces to use RunnableCommand.

### Why are the changes needed?
RunnableCommand is the latest way to run commands.  The advantage over the old way is that we have a single node (no need for a logicalPlan and physical exec node).

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Existing unit tests

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #50393 from szehon-ho/show-namespaces-refactor.

Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: szehon-ho

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.analysis
 
-import org.apache.spark.sql.catalyst.plans.logical.{DescribeNamespace, LogicalPlan, ShowNamespaces, ShowTableProperties, ShowTables}
+import org.apache.spark.sql.catalyst.plans.logical.{DescribeNamespace, LogicalPlan, ShowTableProperties, ShowTables}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.COMMAND
 import org.apache.spark.sql.internal.SQLConf
@@ -36,9 +36,6 @@ object KeepLegacyOutputs extends Rule[LogicalPlan] {
           assert(s.output.length == 3)
           val newOutput = s.output.head.withName("database") +: s.output.tail
           s.copy(output = newOutput)
-        case s: ShowNamespaces =>
-          assert(s.output.length == 1)
-          s.copy(output = Seq(s.output.head.withName("databaseName")))
         case d: DescribeNamespace =>
           assert(d.output.length == 2)
           d.copy(output = Seq(d.output.head.withName("database_description_item"),

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -4408,16 +4408,6 @@ class AstBuilder extends DataTypeAstBuilder
     }
   }
 
-  /**
-   * Create a [[ShowNamespaces]] command.
-   */
-  override def visitShowNamespaces(ctx: ShowNamespacesContext): LogicalPlan = withOrigin(ctx) {
-    val multiPart = Option(ctx.multipartIdentifier).map(visitMultipartIdentifier)
-    ShowNamespaces(
-      UnresolvedNamespace(multiPart.getOrElse(Seq.empty[String])),
-      Option(ctx.pattern).map(x => string(visitStringLit(x))))
-  }
-
   /**
    * Create a [[DescribeNamespace]].
    *

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala

@@ -656,24 +656,6 @@ case class SetNamespaceLocation(
     copy(namespace = newChild)
 }
 
-/**
- * The logical plan of the SHOW NAMESPACES command.
- */
-case class ShowNamespaces(
-    namespace: LogicalPlan,
-    pattern: Option[String],
-    override val output: Seq[Attribute] = ShowNamespaces.getOutputAttrs) extends UnaryCommand {
-  override def child: LogicalPlan = namespace
-  override protected def withNewChildInternal(newChild: LogicalPlan): ShowNamespaces =
-    copy(namespace = newChild)
-}
-
-object ShowNamespaces {
-  def getOutputAttrs: Seq[Attribute] = {
-    Seq(AttributeReference("namespace", StringType, nullable = false)())
-  }
-}
-
 /**
  * The logical plan of the DESCRIBE relation_name command.
  */

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -6687,6 +6687,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def legacyEvalCurrentTime: Boolean = getConf(SQLConf.LEGACY_EVAL_CURRENT_TIME)
 
+  def legacyOutputSchema: Boolean = getConf(SQLConf.LEGACY_KEEP_COMMAND_OUTPUT_SCHEMA)
+
   /** ********************** SQLConf functionality methods ************ */
 
   /** Set Spark SQL configuration properties. */

sql/core/src/main/scala/org/apache/spark/sql/classic/Catalog.scala

@@ -30,13 +30,13 @@ import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.expressions.{Expression, Literal}
 import org.apache.spark.sql.catalyst.parser.ParseException
-import org.apache.spark.sql.catalyst.plans.logical.{ColumnDefinition, CreateTable, LocalRelation, LogicalPlan, OptionList, RecoverPartitions, ShowFunctions, ShowNamespaces, ShowTables, UnresolvedTableSpec, View}
+import org.apache.spark.sql.catalyst.plans.logical.{ColumnDefinition, CreateTable, LocalRelation, LogicalPlan, OptionList, RecoverPartitions, ShowFunctions, ShowTables, UnresolvedTableSpec, View}
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.connector.catalog.{CatalogManager, SupportsNamespaces, TableCatalog}
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.{CatalogHelper, MultipartIdentifierHelper, NamespaceHelper, TransformHelper}
 import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.command.ShowTablesCommand
+import org.apache.spark.sql.execution.command.{ShowNamespacesCommand, ShowTablesCommand}
 import org.apache.spark.sql.execution.datasources.{DataSource, LogicalRelation}
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
 import org.apache.spark.sql.internal.connector.V1Function
@@ -105,10 +105,10 @@ class Catalog(sparkSession: SparkSession) extends catalog.Catalog {
     listDatabasesInternal(Some(pattern))
 
   private def listDatabasesInternal(patternOpt: Option[String]): Dataset[Database] = {
-    val plan = ShowNamespaces(UnresolvedNamespace(Nil), patternOpt)
+    val plan = ShowNamespacesCommand(UnresolvedNamespace(Nil), patternOpt)
     val qe = sparkSession.sessionState.executePlan(plan)
     val catalog = qe.analyzed.collectFirst {
-      case ShowNamespaces(r: ResolvedNamespace, _, _) => r.catalog
+      case ShowNamespacesCommand(r: ResolvedNamespace, _, _) => r.catalog
     }.get
     val databases = qe.toRdd.collect().map { row =>
       // dbName can either be a quoted identifier (single or multi part) or an unquoted single part

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

@@ -1188,4 +1188,11 @@ class SparkSqlAstBuilder extends AstBuilder {
     }
     ShowProceduresCommand(ns)
   }
+
+  override def visitShowNamespaces(ctx: ShowNamespacesContext): LogicalPlan = withOrigin(ctx) {
+    val multiPart = Option(ctx.multipartIdentifier).map(visitMultipartIdentifier)
+    ShowNamespacesCommand(
+      UnresolvedNamespace(multiPart.getOrElse(Seq.empty[String])),
+      Option(ctx.pattern).map(x => string(visitStringLit(x))))
+  }
 }

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

@@ -0,0 +1,77 @@
+/*
+ * 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 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.CatalogV2Implicits.{CatalogHelper, NamespaceHelper}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StringType
+
+/**
+ * The command for `SHOW NAMESPACES`.
+ */
+case class ShowNamespacesCommand(
+    child: LogicalPlan,
+    pattern: Option[String],
+    override val output: Seq[Attribute] = ShowNamespacesCommand.output)
+  extends UnaryRunnableCommand {
+
+  override def run(sparkSession: SparkSession): Seq[Row] = {
+    val ResolvedNamespace(cat, ns, _) = child
+    val nsCatalog = cat.asNamespaceCatalog
+    val namespaces = if (ns.nonEmpty) {
+      nsCatalog.listNamespaces(ns.toArray)
+    } else {
+      nsCatalog.listNamespaces()
+    }
+
+    // The legacy SHOW DATABASES command does not quote the database names.
+    assert(output.length == 1)
+    val namespaceNames = if (output.head.name == "databaseName"
+      && namespaces.forall(_.length == 1)) {
+      namespaces.map(_.head)
+    } else {
+      namespaces.map(_.quoted)
+    }
+
+    namespaceNames
+      .filter{ns => pattern.forall(StringUtils.filterPattern(Seq(ns), _).nonEmpty)}
+      .map(Row(_))
+      .toSeq
+  }
+
+  override protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan = {
+    copy(child = newChild)
+  }
+}
+
+object ShowNamespacesCommand {
+  def output: Seq[AttributeReference] = {
+    Seq(
+      if (SQLConf.get.legacyOutputSchema) {
+        AttributeReference("databaseName", StringType, nullable = false)()
+      } else {
+        AttributeReference("namespace", StringType, nullable = false)()
+      }
+    )
+  }
+}

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

@@ -411,9 +411,6 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
     case DropNamespace(ResolvedNamespace(catalog, ns, _), ifExists, cascade) =>
       DropNamespaceExec(catalog, ns, ifExists, cascade) :: Nil
 
-    case ShowNamespaces(ResolvedNamespace(catalog, ns, _), pattern, output) =>
-      ShowNamespacesExec(output, catalog.asNamespaceCatalog, ns, pattern) :: Nil
-
     case ShowTables(ResolvedNamespace(catalog, ns, _), pattern, output) =>
       ShowTablesExec(output, catalog.asTableCatalog, ns, pattern) :: Nil
 

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

@@ -15,52 +15,59 @@
  * limitations under the License.
  */
 
-package org.apache.spark.sql.execution.command
+package org.apache.spark.sql.execution
 
 import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, UnresolvedNamespace}
-import org.apache.spark.sql.catalyst.parser.CatalystSqlParser.parsePlan
-import org.apache.spark.sql.catalyst.plans.logical.ShowNamespaces
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.command.ShowNamespacesCommand
 import org.apache.spark.sql.test.SharedSparkSession
 
 class ShowNamespacesParserSuite extends AnalysisTest with SharedSparkSession {
+
+  private lazy val parser = new SparkSqlParser()
+
   private val keywords = Seq("NAMESPACES", "DATABASES", "SCHEMAS")
 
+  private def assertEqual(sqlCommand: String, plan: LogicalPlan): Unit = {
+    comparePlans(parser.parsePlan(sqlCommand), plan)
+  }
+
   test("show namespaces in the current catalog") {
     keywords.foreach { keyword =>
-      comparePlans(
-        parsePlan(s"SHOW $keyword"),
-        ShowNamespaces(UnresolvedNamespace(Seq.empty[String]), None))
+      assertEqual(
+        s"SHOW $keyword",
+        ShowNamespacesCommand(UnresolvedNamespace(Seq.empty[String]), None))
     }
   }
 
   test("show namespaces with a pattern") {
     keywords.foreach { keyword =>
-      comparePlans(
-        parsePlan(s"SHOW $keyword LIKE 'defau*'"),
-        ShowNamespaces(UnresolvedNamespace(Seq.empty[String]), Some("defau*")))
+      assertEqual(
+        s"SHOW $keyword LIKE 'defau*'",
+        ShowNamespacesCommand(UnresolvedNamespace(Seq.empty[String]), Some("defau*")))
       // LIKE can be omitted.
-      comparePlans(
-        parsePlan(s"SHOW $keyword 'defau*'"),
-        ShowNamespaces(UnresolvedNamespace(Seq.empty[String]), Some("defau*")))
+      assertEqual(
+        s"SHOW $keyword 'defau*'",
+        ShowNamespacesCommand(UnresolvedNamespace(Seq.empty[String]), Some("defau*")))
     }
   }
 
   test("show namespaces in/from a namespace") {
     keywords.foreach { keyword =>
-      comparePlans(
-        parsePlan(s"SHOW $keyword FROM testcat.ns1.ns2"),
-        ShowNamespaces(UnresolvedNamespace(Seq("testcat", "ns1", "ns2")), None))
-      comparePlans(
-        parsePlan(s"SHOW $keyword IN testcat.ns1.ns2"),
-        ShowNamespaces(UnresolvedNamespace(Seq("testcat", "ns1", "ns2")), None))
+      assertEqual(
+        s"SHOW $keyword FROM testcat.ns1.ns2",
+        ShowNamespacesCommand(UnresolvedNamespace(Seq("testcat", "ns1", "ns2")), None))
+      assertEqual(
+        s"SHOW $keyword IN testcat.ns1.ns2",
+        ShowNamespacesCommand(UnresolvedNamespace(Seq("testcat", "ns1", "ns2")), None))
     }
   }
 
   test("namespaces by a pattern from another namespace") {
     keywords.foreach { keyword =>
-      comparePlans(
-        parsePlan(s"SHOW $keyword IN testcat.ns1 LIKE '*pattern*'"),
-        ShowNamespaces(UnresolvedNamespace(Seq("testcat", "ns1")), Some("*pattern*")))
+      assertEqual(
+        s"SHOW $keyword IN testcat.ns1 LIKE '*pattern*'",
+        ShowNamespacesCommand(UnresolvedNamespace(Seq("testcat", "ns1")), Some("*pattern*")))
     }
   }
 }