[spark] Do not display an unknown partition statistic as a number

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

@@ -21,6 +21,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
+import org.apache.paimon.spark.util.PartitionStatisticsDisplay
 
 import org.apache.spark.sql.{Row, SparkSession}
 import org.apache.spark.sql.catalyst.analysis.ResolvedPartitionSpec
@@ -92,14 +93,21 @@ case class PaimonShowTablePartitionCommand(
     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(", ")}}")
+      // Omit recognized Paimon statistic parameters with negative numeric values.
+      val reported = metadataMap.filterNot {
+        case (field, value) => PartitionStatisticsDisplay.isUnreported(field, value)
+      }
+      if (reported.nonEmpty) {
+        results.put(
+          "Partition Parameters",
+          s"{${reported.map { case (k, v) => s"$k=$v" }.mkString(", ")}}")
+      }
 
-      val fileSizeInBytes =
-        metadataMap.getOrElse(PartitionStatistics.FIELD_FILE_SIZE_IN_BYTES, "0").toLong
+      // Render missing or negative row counts and byte sizes as unknown instead of zero.
       val recordCount =
-        metadataMap.getOrElse(PartitionStatistics.FIELD_RECORD_COUNT, "0").toLong
+        PartitionStatisticsDisplay.render(metadataMap, PartitionStatistics.FIELD_RECORD_COUNT)
+      val fileSizeInBytes =
+        PartitionStatisticsDisplay.render(metadataMap, PartitionStatistics.FIELD_FILE_SIZE_IN_BYTES)
       results.put("Partition Statistics", s"$recordCount rows, $fileSizeInBytes bytes")
     }
 

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/PartitionStatisticsDisplay.scala

@@ -0,0 +1,54 @@
+/*
+ * 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.paimon.spark.util
+
+import org.apache.paimon.partition.PartitionStatistics
+
+import scala.util.Try
+
+/** Formatting helpers for Paimon partition statistics in Spark display commands. */
+object PartitionStatisticsDisplay {
+
+  /** Label for a statistic or creation time that is not known. */
+  val UNKNOWN: String = "UNKNOWN"
+
+  /** The statistic fields Paimon puts into a Spark partition parameter map. */
+  private val STATISTIC_FIELDS: Set[String] = Set(
+    PartitionStatistics.FIELD_RECORD_COUNT,
+    PartitionStatistics.FIELD_FILE_SIZE_IN_BYTES,
+    PartitionStatistics.FIELD_FILE_COUNT,
+    PartitionStatistics.FIELD_LAST_FILE_CREATION_TIME
+  )
+
+  /** Returns true for a recognized Paimon statistic with a negative numeric value. */
+  def isUnreported(field: String, value: String): Boolean =
+    STATISTIC_FIELDS.contains(field) &&
+      asLong(value).exists(count => !PartitionStatistics.isKnown(count))
+
+  /** Returns the numeric value, or [[UNKNOWN]] when it is absent, nonnumeric, or negative. */
+  def render(parameters: collection.Map[String, String], field: String): String =
+    parameters
+      .get(field)
+      .flatMap(asLong)
+      .filter(count => PartitionStatistics.isKnown(count))
+      .map(_.toString)
+      .getOrElse(UNKNOWN)
+
+  private def asLong(value: String): Option[Long] = Try(value.trim.toLong).toOption
+}

paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/PaimonDescribeTableExec.scala

@@ -22,6 +22,7 @@ import org.apache.paimon.partition.PartitionStatistics
 import org.apache.paimon.spark.SparkTable
 import org.apache.paimon.spark.catalog.SparkBaseCatalog
 import org.apache.paimon.spark.leafnode.PaimonLeafV2CommandExec
+import org.apache.paimon.spark.util.PartitionStatisticsDisplay
 import org.apache.paimon.spark.utils.CatalogUtils.{checkNamespace, toIdentifier}
 
 import org.apache.spark.sql.catalyst.InternalRow
@@ -91,23 +92,44 @@ case class PaimonDescribeTableExec(
     }
     val dummyStorageFormat =
       CatalogStorageFormat(None, None, None, None, compressed = false, Map.empty)
-    val partParameters: Map[String, String] = Map(
-      PartitionStatistics.FIELD_FILE_COUNT -> partition.head.fileCount().toString,
-      PartitionStatistics.FIELD_FILE_SIZE_IN_BYTES -> partition.head.fileSizeInBytes().toString,
-      PartitionStatistics.FIELD_LAST_FILE_CREATION_TIME -> partition.head
-        .lastFileCreationTime()
-        .toString,
-      PartitionStatistics.FIELD_RECORD_COUNT -> partition.head.recordCount().toString
-    )
-    val partStats =
-      CatalogStatistics(partition.head.fileSizeInBytes(), Some(partition.head.recordCount()))
-    CatalogTablePartition(
+    val statistics = partition.head
+    // Include only reported values. Spark omits the "Partition Parameters" row for an empty map.
+    val partParameters: Map[String, String] = Seq(
+      PartitionStatistics.FIELD_FILE_COUNT -> statistics.fileCount(),
+      PartitionStatistics.FIELD_FILE_SIZE_IN_BYTES -> statistics.fileSizeInBytes(),
+      PartitionStatistics.FIELD_LAST_FILE_CREATION_TIME -> statistics.lastFileCreationTime(),
+      PartitionStatistics.FIELD_RECORD_COUNT -> statistics.recordCount()
+    ).collect {
+      case (field, value) if PartitionStatistics.isKnown(value) => field -> value.toString
+    }.toMap
+    // CatalogTablePartition uses this temporary value only to render its "Partition Statistics"
+    // row. It cannot represent an unknown size, so omit the row when size is unknown and omit an
+    // unknown row count independently.
+    val partStats = if (PartitionStatistics.isKnown(statistics.fileSizeInBytes())) {
+      val rowCount =
+        if (PartitionStatistics.isKnown(statistics.recordCount())) {
+          Some(BigInt(statistics.recordCount()))
+        } else {
+          None
+        }
+      Some(CatalogStatistics(statistics.fileSizeInBytes(), rowCount))
+    } else {
+      None
+    }
+    val partitionDetails = CatalogTablePartition(
       partitionSpec,
       dummyStorageFormat,
       partParameters,
-      partition.head.lastFileCreationTime(),
+      statistics.lastFileCreationTime(),
       -1,
-      Some(partStats)).toLinkedHashMap.foreach(s => rows += toCatalystRow(s._1, s._2, ""))
+      partStats).toLinkedHashMap
+    // Spark formats every createTime as a date; replace an unknown timestamp to avoid a 1969 date.
+    if (!PartitionStatistics.isKnown(statistics.lastFileCreationTime())) {
+      partitionDetails.put(
+        PaimonDescribeTableExec.CREATED_TIME_KEY,
+        PartitionStatisticsDisplay.UNKNOWN)
+    }
+    partitionDetails.foreach(s => rows += toCatalystRow(s._1, s._2, ""))
     rows += emptyRow()
   }
 
@@ -124,6 +146,9 @@ case class PaimonDescribeTableExec(
 }
 
 object PaimonDescribeTableExec {
+  // Key for the partition creation-time row emitted by CatalogTablePartition.toLinkedHashMap.
+  val CREATED_TIME_KEY = "Created Time"
+
   // This column metadata indicates the default value associated with a particular table column that
   // is in effect at any given time. Its value begins at the time of the initial CREATE/REPLACE
   // TABLE statement with DEFAULT column definition(s), if any. It then changes whenever an ALTER