[SPARK-39350][SQL] Add flag to control breaking change process for: DESC NAMESPACE EXTENDED should redact properties

### What changes were proposed in this pull request?

Add a flag to control breaking change process for: DESC NAMESPACE EXTENDED should redact properties.

### Why are the changes needed?

This lets Spark users control how the new behavior rolls out to users.

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

No.

### How was this patch tested?

This PR extends unit test coverage.

Closes #36799 from dtenedor/desc-namespace-breaking-change.

Authored-by: Daniel Tenedorio <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: dtenedor

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

@@ -3818,6 +3818,16 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
+ val LEGACY_DESC_NAMESPACE_REDACT_PROPERTIES =
+    buildConf("spark.sql.legacy.descNamespaceRedactProperties")
+      .internal()
+      .doc("When set to false, redact sensitive information in the result of DESC NAMESPACE " +
+        "EXTENDED. If set to true, it restores the legacy behavior that this sensitive " +
+        "information was included in the output.")
+      .version("3.4.0")
+      .booleanConf
+      .createWithDefault(false)
+
   /**
    * Holds information about keys that have been deprecated.
    *

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

@@ -190,6 +190,8 @@ case class DescribeDatabaseCommand(
       val propertiesStr =
         if (properties.isEmpty) {
           ""
+        } else if (SQLConf.get.getConf(SQLConf.LEGACY_DESC_NAMESPACE_REDACT_PROPERTIES)) {
+          properties.toSeq.sortBy(_._1).mkString("(", ", ", ")")
         } else {
           conf.redactOptions(properties).toSeq.sortBy(_._1).mkString("(", ", ", ")")
         }

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

@@ -23,6 +23,7 @@ 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.connector.catalog.{CatalogV2Util, SupportsNamespaces}
+import org.apache.spark.sql.internal.SQLConf
 
 /**
  * Physical plan node for describing a namespace.
@@ -48,6 +49,8 @@ case class DescribeNamespaceExec(
       val propertiesStr =
         if (properties.isEmpty) {
           ""
+        } else if (SQLConf.get.getConf(SQLConf.LEGACY_DESC_NAMESPACE_REDACT_PROPERTIES)) {
+          properties.toSeq.sortBy(_._1).mkString("(", ", ", ")")
         } else {
           conf.redactOptions(properties.toMap).toSeq.sortBy(_._1).mkString("(", ", ", ")")
         }

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

@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.command.v2
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.connector.catalog.SupportsNamespaces
 import org.apache.spark.sql.execution.command
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{BooleanType, MetadataBuilder, StringType, StructType}
 import org.apache.spark.util.Utils
 
@@ -30,28 +31,49 @@ class DescribeNamespaceSuite extends command.DescribeNamespaceSuiteBase with Com
   override def notFoundMsgPrefix: String = "Namespace"
 
   test("DescribeNamespace using v2 catalog") {
-    withNamespace(s"$catalog.ns1.ns2") {
-      sql(
-        s"""
-           | CREATE NAMESPACE IF NOT EXISTS $catalog.ns1.ns2
-           | COMMENT 'test namespace'
-           | LOCATION '/tmp/ns_test'
-           | WITH DBPROPERTIES (password = 'password')
+    withSQLConf(SQLConf.LEGACY_DESC_NAMESPACE_REDACT_PROPERTIES.key -> "false") {
+      withNamespace(s"$catalog.ns1.ns2") {
+        sql(
+          s"""
+             | CREATE NAMESPACE IF NOT EXISTS $catalog.ns1.ns2
+             | COMMENT 'test namespace'
+             | LOCATION '/tmp/ns_test'
+             | WITH DBPROPERTIES (password = 'password')
            """.stripMargin)
-      val descriptionDf = sql(s"DESCRIBE NAMESPACE EXTENDED $catalog.ns1.ns2")
-      assert(descriptionDf.schema.map(field => (field.name, field.dataType)) ===
-        Seq(
-          ("info_name", StringType),
-          ("info_value", StringType)
-        ))
-      val description = descriptionDf.collect()
-      assert(description === Seq(
-        Row("Namespace Name", "ns2"),
-        Row(SupportsNamespaces.PROP_COMMENT.capitalize, "test namespace"),
-        Row(SupportsNamespaces.PROP_LOCATION.capitalize, "file:/tmp/ns_test"),
-        Row(SupportsNamespaces.PROP_OWNER.capitalize, Utils.getCurrentUserName()),
-        Row("Properties", "((password,*********(redacted)))"))
-      )
+        val descriptionDf = sql(s"DESCRIBE NAMESPACE EXTENDED $catalog.ns1.ns2")
+        assert(descriptionDf.schema.map(field => (field.name, field.dataType)) ===
+          Seq(
+            ("info_name", StringType),
+            ("info_value", StringType)
+          ))
+        val description = descriptionDf.collect()
+        assert(description === Seq(
+          Row("Namespace Name", "ns2"),
+          Row(SupportsNamespaces.PROP_COMMENT.capitalize, "test namespace"),
+          Row(SupportsNamespaces.PROP_LOCATION.capitalize, "file:/tmp/ns_test"),
+          Row(SupportsNamespaces.PROP_OWNER.capitalize, Utils.getCurrentUserName()),
+          Row("Properties", "((password,*********(redacted)))"))
+        )
+      }
+    }
+    withSQLConf(SQLConf.LEGACY_DESC_NAMESPACE_REDACT_PROPERTIES.key -> "true") {
+      withNamespace(s"$catalog.ns1.ns2") {
+        sql(s"CREATE NAMESPACE IF NOT EXISTS $catalog.ns1.ns2 COMMENT " +
+          "'test namespace' LOCATION '/tmp/ns_test'")
+        val descriptionDf = sql(s"DESCRIBE NAMESPACE $catalog.ns1.ns2")
+        assert(descriptionDf.schema.map(field => (field.name, field.dataType)) ===
+          Seq(
+            ("info_name", StringType),
+            ("info_value", StringType)
+          ))
+        val description = descriptionDf.collect()
+        assert(description === Seq(
+          Row("Namespace Name", "ns2"),
+          Row(SupportsNamespaces.PROP_COMMENT.capitalize, "test namespace"),
+          Row(SupportsNamespaces.PROP_LOCATION.capitalize, "file:/tmp/ns_test"),
+          Row(SupportsNamespaces.PROP_OWNER.capitalize, Utils.getCurrentUserName()))
+        )
+      }
     }
   }