[api][spark] Treat an unreported partition statistic as unknown, not as zero

paimon-api/src/main/java/org/apache/paimon/partition/Partition.java

@@ -74,20 +74,19 @@ public class Partition extends PartitionStatistics {
     @Nullable
     private final Map<String, String> options;
 
-    @JsonCreator
     public Partition(
-            @JsonProperty(FIELD_SPEC) Map<String, String> spec,
-            @JsonProperty(FIELD_RECORD_COUNT) long recordCount,
-            @JsonProperty(FIELD_FILE_SIZE_IN_BYTES) long fileSizeInBytes,
-            @JsonProperty(FIELD_FILE_COUNT) long fileCount,
-            @JsonProperty(FIELD_LAST_FILE_CREATION_TIME) long lastFileCreationTime,
-            @JsonProperty(FIELD_TOTAL_BUCKETS) int totalBuckets,
-            @JsonProperty(FIELD_DONE) boolean done,
-            @JsonProperty(FIELD_CREATED_AT) @Nullable Long createdAt,
-            @JsonProperty(FIELD_CREATED_BY) @Nullable String createdBy,
-            @JsonProperty(FIELD_UPDATED_AT) @Nullable Long updatedAt,
-            @JsonProperty(FIELD_UPDATED_BY) @Nullable String updatedBy,
-            @JsonProperty(FIELD_OPTIONS) @Nullable Map<String, String> options) {
+            Map<String, String> spec,
+            long recordCount,
+            long fileSizeInBytes,
+            long fileCount,
+            long lastFileCreationTime,
+            int totalBuckets,
+            boolean done,
+            @Nullable Long createdAt,
+            @Nullable String createdBy,
+            @Nullable Long updatedAt,
+            @Nullable String updatedBy,
+            @Nullable Map<String, String> options) {
         super(spec, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime, totalBuckets);
         this.done = done;
         this.createdAt = createdAt;
@@ -120,6 +119,44 @@ public Partition(
                 null);
     }
 
+    /**
+     * Reads a partition off the wire. The statistics are optional in the REST contract, so an
+     * absent one decodes to {@link #UNKNOWN} rather than to {@code 0}. {@code totalBuckets} keeps
+     * its {@code 0} default, for writers older than that field.
+     */
+    @JsonCreator
+    static Partition fromJson(
+            @JsonProperty(FIELD_SPEC) Map<String, String> spec,
+            @JsonProperty(FIELD_RECORD_COUNT) @Nullable Long recordCount,
+            @JsonProperty(FIELD_FILE_SIZE_IN_BYTES) @Nullable Long fileSizeInBytes,
+            @JsonProperty(FIELD_FILE_COUNT) @Nullable Long fileCount,
+            @JsonProperty(FIELD_LAST_FILE_CREATION_TIME) @Nullable Long lastFileCreationTime,
+            @JsonProperty(FIELD_TOTAL_BUCKETS) int totalBuckets,
+            @JsonProperty(FIELD_DONE) boolean done,
+            @JsonProperty(FIELD_CREATED_AT) @Nullable Long createdAt,
+            @JsonProperty(FIELD_CREATED_BY) @Nullable String createdBy,
+            @JsonProperty(FIELD_UPDATED_AT) @Nullable Long updatedAt,
+            @JsonProperty(FIELD_UPDATED_BY) @Nullable String updatedBy,
+            @JsonProperty(FIELD_OPTIONS) @Nullable Map<String, String> options) {
+        return new Partition(
+                spec,
+                orUnknown(recordCount),
+                orUnknown(fileSizeInBytes),
+                orUnknown(fileCount),
+                orUnknown(lastFileCreationTime),
+                totalBuckets,
+                done,
+                createdAt,
+                createdBy,
+                updatedAt,
+                updatedBy,
+                options);
+    }
+
+    private static long orUnknown(@Nullable Long value) {
+        return value == null ? UNKNOWN : value;
+    }
+
     @JsonGetter(FIELD_DONE)
     public boolean done() {
         return done;

paimon-api/src/test/java/org/apache/paimon/partition/PartitionTest.java

@@ -61,6 +61,65 @@ void testJsonSerializationWithNullValues() {
         assertThat(json).contains("totalBuckets");
     }
 
+    @Test
+    void testAbsentStatisticsAreUnknownNotZero() {
+        // What listPartitions returns from a catalog that stores no statistics. Every consumer of
+        // this response reads the numbers through PartitionStatistics.isKnown, so absence has to
+        // arrive as unknown and not as an exact zero.
+        String statisticsFreeJson = "{\"spec\":{\"pt\":\"1\"},\"done\":true}";
+
+        Partition partition = JsonSerdeUtil.fromJson(statisticsFreeJson, Partition.class);
+
+        assertThat(partition.spec()).containsEntry("pt", "1");
+        assertThat(partition.done()).isTrue();
+        assertThat(partition.recordCount()).isEqualTo(PartitionStatistics.UNKNOWN);
+        assertThat(partition.fileSizeInBytes()).isEqualTo(PartitionStatistics.UNKNOWN);
+        assertThat(partition.fileCount()).isEqualTo(PartitionStatistics.UNKNOWN);
+        assertThat(partition.lastFileCreationTime()).isEqualTo(PartitionStatistics.UNKNOWN);
+        assertThat(partition.createdAt()).isNull();
+        assertThat(partition.options()).isNull();
+    }
+
+    @Test
+    void testReportedZeroStaysAnExactMeasurement() {
+        // The other half of the same boundary: a partition someone measured as empty must not come
+        // back as unknown.
+        String emptyPartitionJson =
+                "{\"spec\":{\"pt\":\"1\"},\"recordCount\":0,\"fileSizeInBytes\":0,"
+                        + "\"fileCount\":0,\"lastFileCreationTime\":0}";
+
+        Partition partition = JsonSerdeUtil.fromJson(emptyPartitionJson, Partition.class);
+
+        assertThat(partition.recordCount()).isEqualTo(0L);
+        assertThat(partition.fileSizeInBytes()).isEqualTo(0L);
+        assertThat(partition.fileCount()).isEqualTo(0L);
+        assertThat(PartitionStatistics.isKnown(partition.recordCount())).isTrue();
+    }
+
+    @Test
+    void testMeasurementsSurviveARoundTrip() {
+        Partition partition =
+                new Partition(
+                        Collections.singletonMap("pt", "1"),
+                        0L, // an empty partition someone did measure
+                        1024L,
+                        2L,
+                        1234567890L,
+                        10,
+                        true,
+                        1234567890L,
+                        "user1",
+                        1234567900L,
+                        "user2",
+                        Collections.singletonMap("key", "value"));
+
+        Partition parsed =
+                JsonSerdeUtil.fromJson(JsonSerdeUtil.toFlatJson(partition), Partition.class);
+
+        assertThat(parsed).isEqualTo(partition);
+        assertThat(parsed.recordCount()).isEqualTo(0L);
+    }
+
     @Test
     void testJsonSerializationWithNonNullValues() {
         Map<String, String> spec = Collections.singletonMap("pt", "1");

paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java

@@ -37,6 +37,7 @@
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.JsonSerdeUtil;
 import org.apache.paimon.utils.PartitionPathUtils;
 
 import org.junit.jupiter.api.DisplayName;
@@ -206,6 +207,31 @@ void testPlanReusesCatalogListingForSplitsAndRowCount() throws Exception {
         verify(catalog).listPartitionsPaged(IDENTIFIER, 1000, null, null);
     }
 
+    @Test
+    void testPlanRowCountStaysUnknownWhenCatalogReportsNoStatistics() throws Exception {
+        // Partitions as they come off the wire from a catalog that stores no statistics. Summing
+        // them as zeros would tell Spark the table is empty and get a huge scan broadcast.
+        Catalog catalog = mock(Catalog.class);
+        Partition october =
+                JsonSerdeUtil.fromJson(
+                        "{\"spec\":{\"year\":\"2025\",\"month\":\"10\"}}", Partition.class);
+        Partition november =
+                JsonSerdeUtil.fromJson(
+                        "{\"spec\":{\"year\":\"2025\",\"month\":\"11\"}}", Partition.class);
+        when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+                .thenReturn(new PagedList<>(Arrays.asList(october, november), null));
+        LocalFileIO fileIO = LocalFileIO.create();
+        Path tablePath = new Path(tempDir.toUri());
+        writeDataFile(fileIO, tablePath, "year=2025/month=10");
+        writeDataFile(fileIO, tablePath, "year=2025/month=11");
+        FormatTable table = createTable(fileIO, tablePath, partitionManager(catalog), false);
+
+        FormatTableScan.Plan plan = new FormatTableScan(table, null, null).plan();
+
+        assertThat(plan.splits()).hasSize(2);
+        assertThat(plan.rowCount()).isEqualTo(OptionalLong.empty());
+    }
+
     @Test
     void testUnderscoreInPartitionNameRemainsLiteralPrefix() {
         LocalFileIO fileIO = LocalFileIO.create();

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/PaimonStatistics.scala

@@ -43,9 +43,17 @@ case class PaimonStatistics(
     scanRowCount: OptionalLong = OptionalLong.empty()
 ) extends Statistics {
 
+  private lazy val fileTotalSize: Long = splits.map(SplitUtils.splitSize).sum
+
   lazy val numRows: OptionalLong = {
     if (scanRowCount.isPresent) {
-      scanRowCount
+      // A catalog may report zero because it cannot tell "never measured" from "measured, and
+      // empty". Over real bytes leave it unknown; over zero bytes the scan really is empty.
+      if (scanRowCount.getAsLong > 0 || fileTotalSize == 0) {
+        scanRowCount
+      } else {
+        OptionalLong.empty()
+      }
     } else if (splits.exists(_.rowCount() == -1)) {
       OptionalLong.empty()
     } else {
@@ -54,20 +62,19 @@ case class PaimonStatistics(
   }
 
   lazy val sizeInBytes: OptionalLong = {
-    if (numRows.isPresent) {
+    if (numRows.isPresent && numRows.getAsLong > 0) {
       val sizeInBytes = numRows.getAsLong * estimateRowSize(readRowType)
       // Avoid return 0 bytes if there are some valid rows.
       // Avoid return too small size in bytes which may less than row count,
       // note the compression ratio on disk is usually bigger than memory.
       OptionalLong.of(Math.max(sizeInBytes, numRows.getAsLong))
+    } else if (fileTotalSize > 0) {
+      // Zero rows times any row size is zero, so weigh the files instead.
+      OptionalLong.of((fileTotalSize * readRowSizeRatio).toLong)
+    } else if (numRows.isPresent) {
+      OptionalLong.of(0L)
     } else {
-      val fileTotalSize = splits.map(SplitUtils.splitSize).sum
-      if (fileTotalSize == 0) {
-        OptionalLong.empty()
-      } else {
-        val size = (fileTotalSize * readRowSizeRatio).toLong
-        OptionalLong.of(size)
-      }
+      OptionalLong.empty()
     }
   }
 

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

@@ -376,6 +376,45 @@ class CatalogManagedPartitionAnalyzeTest extends PaimonSparkTestWithRestCatalogB
     }
   }
 
+  test("a zero row count does not make a partition that still has files look free to read") {
+    val tableName = "analyze_zero_row_count_size"
+    withTable(tableName) {
+      createTable(tableName)
+      sql(
+        s"INSERT INTO ${qualified(tableName)} VALUES " +
+          s"(1, 'a', '20260101', '00'), (2, 'b', '20260101', '00')")
+
+      // Positive control: a real measurement gives both a row count and a size.
+      val measured = getFormatTableScan(s"SELECT * FROM ${qualified(tableName)}").estimateStatistics
+      assert(measured.numRows().getAsLong == 2L)
+      assert(measured.sizeInBytes().getAsLong > 0L)
+
+      // Rewrite the statistics to zero without touching the files, the way a catalog that cannot
+      // tell "never measured" from "measured, and empty" answers.
+      val spec = Map("dt" -> "20260101", "hour" -> "00").asJava
+      paimonCatalog.createPartitions(
+        Identifier.create(dbName0, tableName),
+        List(spec).asJava,
+        true,
+        List(
+          new PartitionStatistics(
+            spec,
+            0L,
+            0L,
+            0L,
+            0L,
+            PartitionStatistics.UNKNOWN_TOTAL_BUCKETS)).asJava,
+        true
+      )
+
+      val zeroed = getFormatTableScan(s"SELECT * FROM ${qualified(tableName)}").estimateStatistics
+      // Neither number may say the scan is free: the size drives broadcast, the row count drives
+      // join reordering.
+      assert(!zeroed.numRows().isPresent, zeroed.numRows().toString)
+      assert(zeroed.sizeInBytes().getAsLong > 0L, zeroed.sizeInBytes().toString)
+    }
+  }
+
   test("partition row count is not duplicated across format data splits") {
     val tableName = "analyze_multi_split_statistics"
     withTable(tableName) {