[SPARK-37150][SQL] Migrate DESCRIBE NAMESPACE to use V2 command by default

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.{LogicalPlan, ShowNamespaces, ShowTables}
+import org.apache.spark.sql.catalyst.plans.logical.{DescribeNamespace, LogicalPlan, ShowNamespaces, 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
@@ -39,6 +39,10 @@ object KeepLegacyOutputs extends Rule[LogicalPlan] {
         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"),
+            d.output.last.withName("database_description_value")))
       }
     }
   }

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

@@ -105,15 +105,8 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
     case UnsetViewProperties(ResolvedView(ident, _), keys, ifExists) =>
       AlterTableUnsetPropertiesCommand(ident.asTableIdentifier, keys, ifExists, isView = true)
 
-    case DescribeNamespace(DatabaseInSessionCatalog(db), extended, output) =>
-      val newOutput = if (conf.getConf(SQLConf.LEGACY_KEEP_COMMAND_OUTPUT_SCHEMA)) {
-        assert(output.length == 2)
-        Seq(output.head.withName("database_description_item"),
-          output.last.withName("database_description_value"))
-      } else {
-        output
-      }
-      DescribeDatabaseCommand(db, extended, newOutput)
+    case DescribeNamespace(DatabaseInSessionCatalog(db), extended, output) if conf.useV1Command =>
+      DescribeDatabaseCommand(db, extended, output)
 
     case SetNamespaceProperties(DatabaseInSessionCatalog(db), properties) =>
       AlterDatabasePropertiesCommand(db, properties)

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

@@ -44,10 +44,13 @@ case class DescribeNamespaceExec(
     }
 
     if (isExtended) {
-      val properties = metadata.asScala -- CatalogV2Util.NAMESPACE_RESERVED_PROPERTIES
-      if (properties.nonEmpty) {
-        rows += toCatalystRow("Properties", properties.toSeq.mkString("(", ",", ")"))
+      val properties = metadata.asScala.toMap -- CatalogV2Util.NAMESPACE_RESERVED_PROPERTIES
+      val propertiesStr = if (properties.isEmpty) {
+        ""
+      } else {
+        properties.toSeq.mkString("(", ", ", ")")
       }
+      rows += toCatalystRow("Properties", propertiesStr)
     }
     rows.toSeq
   }

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala

@@ -1079,7 +1079,8 @@ class DataSourceV2SQLSuite
           assert(sql("DESC NAMESPACE EXTENDED testcat.reservedTest")
             .toDF("k", "v")
             .where("k='Properties'")
-            .isEmpty, s"$key is a reserved namespace property and ignored")
+            .where("v=''")
+            .count == 1, s"$key is a reserved namespace property and ignored")
           val meta =
             catalog("testcat").asNamespaceCatalog.loadNamespaceMetadata(Array("reservedTest"))
           assert(meta.get(key) == null || !meta.get(key).contains("foo"),
@@ -1244,7 +1245,7 @@ class DataSourceV2SQLSuite
         Row(SupportsNamespaces.PROP_COMMENT.capitalize, "test namespace"),
         Row(SupportsNamespaces.PROP_LOCATION.capitalize, "/tmp/ns_test"),
         Row(SupportsNamespaces.PROP_OWNER.capitalize, defaultUser),
-        Row("Properties", "((a,b),(b,a),(c,c))"))
+        Row("Properties", "((a,b), (b,a), (c,c))"))
       )
     }
   }
@@ -1270,7 +1271,8 @@ class DataSourceV2SQLSuite
           assert(sql("DESC NAMESPACE EXTENDED testcat.reservedTest")
             .toDF("k", "v")
             .where("k='Properties'")
-            .isEmpty, s"$key is a reserved namespace property and ignored")
+            .where("v=''")
+            .count == 1, s"$key is a reserved namespace property and ignored")
           val meta =
             catalog("testcat").asNamespaceCatalog.loadNamespaceMetadata(Array("reservedTest"))
           assert(meta.get(key) == null || !meta.get(key).contains("foo"),
@@ -1290,7 +1292,8 @@ class DataSourceV2SQLSuite
         Row("Namespace Name", "ns2"),
         Row(SupportsNamespaces.PROP_COMMENT.capitalize, "test namespace"),
         Row(SupportsNamespaces.PROP_LOCATION.capitalize, "/tmp/ns_test_2"),
-        Row(SupportsNamespaces.PROP_OWNER.capitalize, defaultUser))
+        Row(SupportsNamespaces.PROP_OWNER.capitalize, defaultUser),
+        Row("Properties", ""))
       )
     }
   }

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

@@ -762,7 +762,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
         checkAnswer(
           sql(s"DESCRIBE DATABASE EXTENDED $dbName").toDF("key", "value")
             .where("key not like 'Owner%'"), // filter for consistency with in-memory catalog
-          Row("Database Name", dbNameWithoutBackTicks) ::
+          Row("Namespace Name", dbNameWithoutBackTicks) ::
             Row("Comment", "") ::
             Row("Location", CatalogUtils.URIToString(location)) ::
             Row("Properties", "((a,a), (b,b), (c,c))") :: Nil)
@@ -772,7 +772,7 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
         checkAnswer(
           sql(s"DESCRIBE DATABASE EXTENDED $dbName").toDF("key", "value")
             .where("key not like 'Owner%'"), // filter for consistency with in-memory catalog
-          Row("Database Name", dbNameWithoutBackTicks) ::
+          Row("Namespace Name", dbNameWithoutBackTicks) ::
             Row("Comment", "") ::
             Row("Location", CatalogUtils.URIToString(location)) ::
             Row("Properties", "((a,a), (b,b), (c,c), (d,d))") :: Nil)

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.execution.command
 
 import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.internal.SQLConf
 
 /**
  * This base suite contains unified tests for the `DESCRIBE NAMESPACE` command that check V1 and V2
@@ -47,4 +48,21 @@ trait DescribeNamespaceSuiteBase extends QueryTest with DDLCommandTestUtils {
     // TODO: Move this to DropNamespaceSuite when the test suite is introduced.
     sql(s"DROP NAMESPACE IF EXISTS $catalog.$ns")
   }
+
+  test("Keep the legacy output schema") {
+    Seq(true, false).foreach { keepLegacySchema =>
+      withSQLConf(SQLConf.LEGACY_KEEP_COMMAND_OUTPUT_SCHEMA.key -> keepLegacySchema.toString) {
+        val ns = "db1"
+        withNamespace(ns) {
+          sql(s"CREATE NAMESPACE $ns")
+          val schema = sql(s"DESCRIBE NAMESPACE $ns").schema.fieldNames.toSeq
+          if (keepLegacySchema) {
+            assert(schema === Seq("database_description_item", "database_description_value"))
+          } else {
+            assert(schema === Seq("info_name", "info_value"))
+          }
+        }
+      }
+    }
+  }
 }

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

@@ -19,7 +19,6 @@ package org.apache.spark.sql.execution.command.v1
 
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.execution.command
-import org.apache.spark.sql.internal.SQLConf
 
 /**
  * This base suite contains unified tests for the `DESCRIBE NAMESPACE` command that checks V1
@@ -30,7 +29,8 @@ import org.apache.spark.sql.internal.SQLConf
  *   - V1 Hive External catalog:
  *     `org.apache.spark.sql.hive.execution.command.DescribeNamespaceSuite`
  */
-trait DescribeNamespaceSuiteBase extends command.DescribeNamespaceSuiteBase {
+trait DescribeNamespaceSuiteBase extends command.DescribeNamespaceSuiteBase
+    with command.TestsV1AndV2Commands {
   override def notFoundMsgPrefix: String = "Database"
 
   test("basic") {
@@ -43,35 +43,21 @@ trait DescribeNamespaceSuiteBase extends command.DescribeNamespaceSuiteBase {
         .where("key not like 'Owner%'") // filter for consistency with in-memory catalog
         .collect()
 
+      val namePrefix = if (conf.useV1Command) "Database" else "Namespace"
       assert(result.length == 4)
-      assert(result(0) === Row("Database Name", ns))
+      assert(result(0) === Row(s"$namePrefix Name", ns))
       assert(result(1) === Row("Comment", ""))
       // Check only the key for "Location" since its value depends on warehouse path, etc.
       assert(result(2).getString(0) === "Location")
       assert(result(3) === Row("Properties", ""))
     }
   }
-
-  test("Keep the legacy output schema") {
-    Seq(true, false).foreach { keepLegacySchema =>
-      withSQLConf(SQLConf.LEGACY_KEEP_COMMAND_OUTPUT_SCHEMA.key -> keepLegacySchema.toString) {
-        val ns = "db1"
-        withNamespace(ns) {
-          sql(s"CREATE NAMESPACE $ns")
-          val schema = sql(s"DESCRIBE NAMESPACE $ns").schema.fieldNames.toSeq
-          if (keepLegacySchema) {
-            assert(schema === Seq("database_description_item", "database_description_value"))
-          } else {
-            assert(schema === Seq("info_name", "info_value"))
-          }
-        }
-      }
-    }
-  }
 }
 
 /**
  * The class contains tests for the `DESCRIBE NAMESPACE` command to check V1 In-Memory
  * table catalog.
  */
-class DescribeNamespaceSuite extends DescribeNamespaceSuiteBase with CommandSuiteBase
+class DescribeNamespaceSuite extends DescribeNamespaceSuiteBase with CommandSuiteBase {
+  override def commandVersion: String = super[DescribeNamespaceSuiteBase].commandVersion
+}

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/DescribeNamespaceSuite.scala

@@ -23,4 +23,6 @@ import org.apache.spark.sql.execution.command.v1
  * The class contains tests for the `DESCRIBE NAMESPACE` command to check V1 Hive external
  * table catalog.
  */
-class DescribeNamespaceSuite extends v1.DescribeNamespaceSuiteBase with CommandSuiteBase
+class DescribeNamespaceSuite extends v1.DescribeNamespaceSuiteBase with CommandSuiteBase {
+  override def commandVersion: String = super[DescribeNamespaceSuiteBase].commandVersion
+}