[spark] Support partition statistics in SHOW TABLE EXTENDED PARTITION command

Author: kerwin-zk

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala

@@ -19,7 +19,9 @@
 package org.apache.paimon.spark
 
 import org.apache.paimon.CoreOptions
+import org.apache.paimon.partition.PartitionStatistics
 import org.apache.paimon.table.{FileStoreTable, Table}
+import org.apache.paimon.table.source.ScanMode
 import org.apache.paimon.types.RowType
 import org.apache.paimon.utils.{InternalRowPartitionComputer, TypeUtils}
 
@@ -136,7 +138,31 @@ trait PaimonPartitionManagement extends SupportsAtomicPartitionManagement with L
   }
 
   override def loadPartitionMetadata(ident: InternalRow): JMap[String, String] = {
-    Map.empty[String, String].asJava
+    table match {
+      case fileStoreTable: FileStoreTable =>
+        val partitionSpec = toPaimonPartitions(Array(ident)).head
+        val partitionEntries = fileStoreTable
+          .newSnapshotReader()
+          .withMode(ScanMode.ALL)
+          .withPartitionFilter(partitionSpec)
+          .partitionEntries()
+
+        if (!partitionEntries.isEmpty) {
+          val entry = partitionEntries.get(0)
+          Map(
+            PartitionStatistics.FIELD_RECORD_COUNT -> entry.recordCount().toString,
+            PartitionStatistics.FIELD_FILE_SIZE_IN_BYTES -> entry.fileSizeInBytes().toString,
+            PartitionStatistics.FIELD_FILE_COUNT -> entry.fileCount().toString,
+            PartitionStatistics.FIELD_LAST_FILE_CREATION_TIME -> entry
+              .lastFileCreationTime()
+              .toString
+          ).asJava
+        } else {
+          Map.empty[String, String].asJava
+        }
+      case _ =>
+        Map.empty[String, String].asJava
+    }
   }
 
   override def listPartitionIdentifiers(

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DescribeTableTestBase.scala

@@ -98,6 +98,36 @@ abstract class DescribeTableTestBase extends PaimonSparkTestBase {
           )
           Assertions.assertTrue(
             res2.select("information").collect().head.getString(0).contains("Partition Values"))
+
+          val info2 = res2.select("information").collect().head.getString(0)
+          Assertions.assertTrue(
+            info2.contains("Partition Parameters"),
+            s"SHOW TABLE EXTENDED should contain Partition Parameters, but got: $info2")
+          Assertions.assertTrue(
+            info2.contains(PartitionStatistics.FIELD_RECORD_COUNT),
+            s"SHOW TABLE EXTENDED should contain recordCount, but got: $info2")
+          Assertions.assertTrue(
+            info2.contains(PartitionStatistics.FIELD_FILE_SIZE_IN_BYTES),
+            s"SHOW TABLE EXTENDED should contain fileSizeInBytes, but got: $info2")
+          Assertions.assertTrue(
+            info2.contains(PartitionStatistics.FIELD_FILE_COUNT),
+            s"SHOW TABLE EXTENDED should contain fileCount, but got: $info2")
+          Assertions.assertTrue(
+            info2.contains(PartitionStatistics.FIELD_LAST_FILE_CREATION_TIME),
+            s"SHOW TABLE EXTENDED should contain lastFileCreationTime, but got: $info2")
+          // Verify recordCount value: partition (pt1='2024', pt2='11') has 1 row
+          // Spark 3 uses "key=value" format, Spark 4 uses "key: value" format
+          Assertions.assertTrue(
+            info2.contains("recordCount=1") || info2.contains("recordCount: 1"),
+            s"Partition (pt1='2024', pt2='11') should have recordCount=1, but got: $info2")
+
+          // Check s2 partition with known record count
+          val res3 =
+            spark.sql(s"SHOW TABLE EXTENDED IN $testDB LIKE 's2' PARTITION(pt = '2024')")
+          val info3 = res3.select("information").collect().head.getString(0)
+          Assertions.assertTrue(
+            info3.contains("recordCount=2") || info3.contains("recordCount: 2"),
+            s"Partition pt='2024' should have recordCount=2, but got: $info3")
         }
       }
     }

paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/commands/PaimonShowTablePartitionCommand.scala

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.spark.commands
 
+import org.apache.paimon.partition.PartitionStatistics
 import org.apache.paimon.spark.catalyst.Compatibility
 import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
 
@@ -87,7 +88,20 @@ case class PaimonShowTablePartitionCommand(
     val partitionValues = partitions.mkString("[", ", ", "]")
     results.put("Partition Values", s"$partitionValues")
 
-    // TODO "Partition Parameters", "Created Time", "Last Access", "Partition Statistics"
+    // Partition Parameters and Partition Statistics
+    val metadata = partitionTable.loadPartitionMetadata(row)
+    if (!metadata.isEmpty) {
+      val metadataMap = metadata.asScala
+      results.put(
+        "Partition Parameters",
+        s"{${metadataMap.map { case (k, v) => s"$k=$v" }.mkString(", ")}}")
+
+      val fileSizeInBytes =
+        metadataMap.getOrElse(PartitionStatistics.FIELD_FILE_SIZE_IN_BYTES, "0").toLong
+      val recordCount =
+        metadataMap.getOrElse(PartitionStatistics.FIELD_RECORD_COUNT, "0").toLong
+      results.put("Partition Statistics", s"$recordCount rows, $fileSizeInBytes bytes")
+    }
 
     results
       .map {