[SPARK-56613][CORE] Supports displaying all partition parameters

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

@@ -7217,6 +7217,15 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val SHOW_ALL_PARTITION_PARAMETERS_ENABLED =
+    buildConf("spark.sql.hive.showAllPartitionParameters")
+      .doc("When true, Spark does not hide internal `spark.sql.*` entries from " +
+        "partition parameters returned by HiveExternalCatalog. " +
+        "When false, these internal entries are filtered out.")
+      .version("4.2.0")
+      .booleanConf
+      .createWithDefault(false)
+
   /**
    * Holds information about keys that have been deprecated.
    *
@@ -8517,6 +8526,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def listaggAllowDistinctCastWithOrder: Boolean = getConf(LISTAGG_ALLOW_DISTINCT_CAST_WITH_ORDER)
 
+  def showAllPartitionParameters: Boolean = getConf(SQLConf.SHOW_ALL_PARTITION_PARAMETERS_ENABLED)
+
   /** ********************** SQLConf functionality methods ************ */
 
   /** Set Spark SQL configuration properties. */

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala

@@ -48,6 +48,7 @@ import org.apache.spark.sql.execution.command.DDLUtils
 import org.apache.spark.sql.execution.datasources.{PartitioningUtils, SourceOptions}
 import org.apache.spark.sql.hive.client.HiveClient
 import org.apache.spark.sql.internal.HiveSerDe
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.StaticSQLConf._
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.util.SchemaUtils
@@ -1300,11 +1301,16 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
     // Note: partition-level statistics were introduced in 2.3.
     val restoredStats = statsFromProperties(partition.parameters, table.identifier.table)
     if (restoredStats.isDefined) {
+      val filteredParameters = if (SQLConf.get.showAllPartitionParameters) {
+        partition.parameters
+      } else {
+        partition.parameters.filterNot {
+          case (key, _) => key.startsWith(SPARK_SQL_PREFIX) }
+      }
       partition.copy(
         spec = restoredSpec,
         stats = restoredStats,
-        parameters = partition.parameters.filterNot {
-          case (key, _) => key.startsWith(SPARK_SQL_PREFIX) })
+        parameters = filteredParameters)
     } else {
       partition.copy(spec = restoredSpec)
     }

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveExternalCatalogSuite.scala

@@ -26,7 +26,10 @@ import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.execution.QueryExecutionException
 import org.apache.spark.sql.execution.command.DDLUtils
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.SHOW_ALL_PARTITION_PARAMETERS_ENABLED
 import org.apache.spark.sql.types.{StringType, StructField, StructType}
+import org.apache.spark.util.Utils
 
 /**
  * Test suite for the [[HiveExternalCatalog]].
@@ -243,6 +246,61 @@ class HiveExternalCatalogSuite extends ExternalCatalogSuite {
     assert(DataTypeUtils.sameType(alteredTable.schema, newSchema))
   }
 
+  test("partition parameters should be controlled by config") {
+    def createCatalog(): (HiveExternalCatalog, String, Seq[java.io.File]) = {
+      val sparkConf = new SparkConf()
+      val hadoopConf = new Configuration
+      val metastorePath = Utils.createTempDir()
+      metastorePath.delete()
+      val warehousePath = Utils.createTempDir()
+      hadoopConf.set(
+        "javax.jdo.option.ConnectionURL",
+        s"jdbc:derby:;databaseName=${metastorePath.getCanonicalPath};create=true")
+      hadoopConf.set("hive.metastore.warehouse.dir", warehousePath.getCanonicalPath)
+      val catalog = new HiveExternalCatalog(sparkConf, hadoopConf)
+      val db = "db_show_part_params"
+      catalog.client.reset()
+      catalog.createDatabase(newDb(db), ignoreIfExists = false)
+      catalog.createTable(newTable("tbl", db), ignoreIfExists = false)
+
+      val partSpec = Map("a" -> "1", "b" -> "2")
+      val part = CatalogTablePartition(partSpec, storageFormat)
+      catalog.createPartitions(db, "tbl", Seq(part), ignoreIfExists = false)
+
+      val sparkSqlParamKey = HiveExternalCatalog.SPARK_SQL_PREFIX + "test"
+      val updatedPart = catalog.getPartition(db, "tbl", partSpec).copy(
+        parameters = Map("k1" -> "v1", sparkSqlParamKey -> "v2"),
+        stats = Some(CatalogStatistics(sizeInBytes = 1, rowCount = Some(1))))
+      catalog.alterPartitions(db, "tbl", Seq(updatedPart))
+      (catalog, db, Seq(metastorePath, warehousePath))
+    }
+
+    val (catalog, db, cleanupDirs) = createCatalog()
+    try {
+      val hiddenParamsConf = new SQLConf
+      hiddenParamsConf.setConf(SHOW_ALL_PARTITION_PARAMETERS_ENABLED, false)
+      SQLConf.withExistingConf(hiddenParamsConf) {
+        val hiddenParamsPart = catalog.getPartition(db, "tbl", Map("a" -> "1", "b" -> "2"))
+        assert(hiddenParamsPart.stats.nonEmpty)
+        assert(hiddenParamsPart.parameters.get("k1").contains("v1"))
+        assert(hiddenParamsPart.parameters.keys.forall(
+          !_.startsWith(HiveExternalCatalog.SPARK_SQL_PREFIX)))
+      }
+
+      val allParamsConf = new SQLConf
+      allParamsConf.setConf(SHOW_ALL_PARTITION_PARAMETERS_ENABLED, true)
+      SQLConf.withExistingConf(allParamsConf) {
+        val allParamsPart = catalog.getPartition(db, "tbl", Map("a" -> "1", "b" -> "2"))
+        assert(allParamsPart.stats.nonEmpty)
+        assert(allParamsPart.parameters.get("k1").contains("v1"))
+        assert(allParamsPart.parameters.get(HiveExternalCatalog.SPARK_SQL_PREFIX + "test")
+          .contains("v2"))
+      }
+    } finally {
+      cleanupDirs.foreach(Utils.deleteRecursively)
+    }
+  }
+
   test("SPARK-50137: Avoid fallback to Hive-incompatible ways on thrift exception") {
     val hadoopConf = new Configuration()
     // Use an unavailable uri to mock client connection timeout.