[core] Carry the row count and byte size a format table writer already counted

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

@@ -30,15 +30,39 @@
 import java.util.Objects;
 
 /**
- * Statistics of a partition, fields inside may be negative, indicating that some data has been
- * removed.
+ * Statistics of a partition.
+ *
+ * <p>The numeric fields are read on two planes, and a negative value means a different thing on
+ * each. Which plane an instance belongs to follows from where it came from, never from the value:
+ *
+ * <ul>
+ *   <li><b>Delta plane</b> — what a commit changed. A negative value is a decrement, and the server
+ *       adds it to what it already holds. This is what a table snapshot commit reports.
+ *   <li><b>Observation plane</b> — what a partition currently holds, as returned by {@code
+ *       listPartitions}. A negative value ({@link #UNKNOWN}) means nobody ever reported that field,
+ *       and {@code 0} means an exact zero. The two are not interchangeable: a consumer that treats
+ *       unknown as zero plans against an empty partition that may hold a billion rows.
+ * </ul>
+ *
+ * <p>Unknown is per field, not per partition: a reporter that only knows the file count leaves the
+ * record count {@link #UNKNOWN} and fills the rest. Use {@link #isKnown(long)} rather than
+ * comparing against {@code -1}; any negative value on the observation plane is unknown.
  */
 @JsonIgnoreProperties(ignoreUnknown = true)
 @Public
 public class PartitionStatistics implements Serializable {
 
     private static final long serialVersionUID = 1L;
 
+    /**
+     * Canonical encoding of "this field was never reported" on the observation plane. Any negative
+     * value carries the same meaning; this is the one to write.
+     */
+    public static final long UNKNOWN = -1L;
+
+    /** Format tables have no buckets, so their bucket count is always unknown. */
+    public static final int UNKNOWN_TOTAL_BUCKETS = -1;
+
     public static final String FIELD_SPEC = "spec";
     public static final String FIELD_RECORD_COUNT = "recordCount";
     public static final String FIELD_FILE_SIZE_IN_BYTES = "fileSizeInBytes";
@@ -82,6 +106,20 @@ public PartitionStatistics(
         this.totalBuckets = totalBuckets;
     }
 
+    /** Statistics of a partition nobody ever reported on: every field {@link #UNKNOWN}. */
+    public static PartitionStatistics unknown(Map<String, String> spec) {
+        return new PartitionStatistics(
+                spec, UNKNOWN, UNKNOWN, UNKNOWN, UNKNOWN, UNKNOWN_TOTAL_BUCKETS);
+    }
+
+    /**
+     * Whether an observation-plane field carries a real measurement. Never apply this to a
+     * delta-plane value, where a negative number is a decrement rather than a missing measurement.
+     */
+    public static boolean isKnown(long value) {
+        return value >= 0;
+    }
+
     @JsonGetter(FIELD_SPEC)
     public Map<String, String> spec() {
         return spec;

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

@@ -22,6 +22,9 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.util.Collections;
+import java.util.Map;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Test for {@link PartitionStatistics}. */
@@ -41,4 +44,33 @@ void testLegacyPartitionStatisticsDeserialization() {
         assertThat(stats.lastFileCreationTime()).isEqualTo(123456789L);
         assertThat(stats.totalBuckets()).isEqualTo(0);
     }
+
+    @Test
+    void testZeroIsAKnownMeasurement() {
+        // The boundary the whole observation-plane contract rests on: an empty partition was
+        // measured, and a consumer that reads its zero as "nobody looked" plans against the wrong
+        // table.
+        assertThat(PartitionStatistics.isKnown(0L)).isTrue();
+        assertThat(PartitionStatistics.isKnown(1L)).isTrue();
+        assertThat(PartitionStatistics.isKnown(Long.MAX_VALUE)).isTrue();
+
+        assertThat(PartitionStatistics.isKnown(PartitionStatistics.UNKNOWN)).isFalse();
+        // Unknown is any negative value, not only the canonical -1.
+        assertThat(PartitionStatistics.isKnown(-2L)).isFalse();
+        assertThat(PartitionStatistics.isKnown(Long.MIN_VALUE)).isFalse();
+    }
+
+    @Test
+    void testUnknownLeavesEveryFieldUnknown() {
+        Map<String, String> spec = Collections.singletonMap("pt", "1");
+
+        PartitionStatistics stats = PartitionStatistics.unknown(spec);
+
+        assertThat(stats.spec()).isEqualTo(spec);
+        assertThat(PartitionStatistics.isKnown(stats.recordCount())).isFalse();
+        assertThat(PartitionStatistics.isKnown(stats.fileSizeInBytes())).isFalse();
+        assertThat(PartitionStatistics.isKnown(stats.fileCount())).isFalse();
+        assertThat(PartitionStatistics.isKnown(stats.lastFileCreationTime())).isFalse();
+        assertThat(PartitionStatistics.isKnown(stats.totalBuckets())).isFalse();
+    }
 }

paimon-core/src/main/java/org/apache/paimon/io/FormatTableRollingFileWriter.java

@@ -45,7 +45,7 @@ public class FormatTableRollingFileWriter implements AutoCloseable {
     private final long targetFileSize;
     private final long targetFileRowNum;
     private final List<FileWriterAbortExecutor> closedWriters;
-    private final List<TwoPhaseOutputStream.Committer> committers;
+    private final List<FormatTableWrittenFile> writtenFiles;
 
     private FormatTableSingleFileWriter currentWriter = null;
     private long recordCount = 0;
@@ -75,7 +75,7 @@ public FormatTableRollingFileWriter(
         this.targetFileSize = targetFileSize;
         this.targetFileRowNum = targetFileRowNum;
         this.closedWriters = new ArrayList<>();
-        this.committers = new ArrayList<>();
+        this.writtenFiles = new ArrayList<>();
     }
 
     public long targetFileSize() {
@@ -116,7 +116,14 @@ private void closeCurrentWriter() throws IOException {
         currentWriter.close();
         closedWriters.add(currentWriter.abortExecutor());
         if (currentWriter.committers() != null) {
-            committers.addAll(currentWriter.committers());
+            // Read the counts off the writer that produced this file: once it is replaced, the
+            // rows it wrote cannot be recovered without reading the file back.
+            long fileRecordCount = currentWriter.recordCount();
+            long fileSizeInBytes = currentWriter.outputBytes();
+            for (TwoPhaseOutputStream.Committer committer : currentWriter.committers()) {
+                writtenFiles.add(
+                        new FormatTableWrittenFile(committer, fileRecordCount, fileSizeInBytes));
+            }
         }
 
         currentWriter = null;
@@ -128,22 +135,24 @@ public void abort() {
             currentWriter.abort();
             currentWriter = null;
         }
-        for (TwoPhaseOutputStream.Committer committer : committers) {
+        for (FormatTableWrittenFile writtenFile : writtenFiles) {
+            TwoPhaseOutputStream.Committer committer = writtenFile.committer();
             try {
                 committer.discard(fileIO);
             } catch (Throwable e) {
                 LOG.warn("Exception occurs when discarding file {}.", committer.targetPath(), e);
             }
         }
-        committers.clear();
+        writtenFiles.clear();
         for (FileWriterAbortExecutor abortExecutor : closedWriters) {
             abortExecutor.abort();
         }
         closedWriters.clear();
     }
 
-    public List<TwoPhaseOutputStream.Committer> committers() {
-        return committers;
+    /** The files this writer produced, each with the rows and bytes it holds. */
+    public List<FormatTableWrittenFile> writtenFiles() {
+        return writtenFiles;
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/io/FormatTableSingleFileWriter.java

@@ -50,6 +50,7 @@ public class FormatTableSingleFileWriter {
     private TwoPhaseOutputStream.Committer committer;
 
     protected long outputBytes;
+    protected long recordCount;
     protected boolean closed;
 
     public FormatTableSingleFileWriter(
@@ -99,6 +100,7 @@ public void write(InternalRow record) throws IOException {
 
         try {
             writer.addElement(record);
+            recordCount++;
         } catch (Throwable e) {
             LOG.warn("Exception occurs when writing file {}. Cleaning up.", path, e);
             abort();
@@ -140,6 +142,22 @@ public List<TwoPhaseOutputStream.Committer> committers() {
         return Lists.newArrayList(committer);
     }
 
+    /** Rows written to this file. Exact, counted as they were written. */
+    public long recordCount() {
+        if (!closed) {
+            throw new RuntimeException("Writer should be closed before getting record count!");
+        }
+        return recordCount;
+    }
+
+    /** Bytes this file holds, taken from the stream position at close. */
+    public long outputBytes() {
+        if (!closed) {
+            throw new RuntimeException("Writer should be closed before getting output bytes!");
+        }
+        return outputBytes;
+    }
+
     public FileWriterAbortExecutor abortExecutor() {
         if (!closed) {
             throw new RuntimeException("Writer should be closed!");

paimon-core/src/main/java/org/apache/paimon/io/FormatTableWrittenFile.java

@@ -0,0 +1,52 @@
+/*
+ * 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.io;
+
+import org.apache.paimon.fs.TwoPhaseOutputStream;
+
+/**
+ * One data file a format table writer finished, with what it holds. The row count and byte size are
+ * counted while writing, so carrying them alongside the committer costs no extra IO and is the only
+ * place they can still be had exactly — after the commit the file is just bytes on a path.
+ */
+public class FormatTableWrittenFile {
+
+    private final TwoPhaseOutputStream.Committer committer;
+    private final long recordCount;
+    private final long fileSizeInBytes;
+
+    public FormatTableWrittenFile(
+            TwoPhaseOutputStream.Committer committer, long recordCount, long fileSizeInBytes) {
+        this.committer = committer;
+        this.recordCount = recordCount;
+        this.fileSizeInBytes = fileSizeInBytes;
+    }
+
+    public TwoPhaseOutputStream.Committer committer() {
+        return committer;
+    }
+
+    public long recordCount() {
+        return recordCount;
+    }
+
+    public long fileSizeInBytes() {
+        return fileSizeInBytes;
+    }
+}

paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemSplitEnumerator.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.partition.PartitionPredicate;
 import org.apache.paimon.partition.PartitionPredicate.MultiplePartitionPredicate;
+import org.apache.paimon.partition.PartitionStatistics;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.table.FormatTable;
 import org.apache.paimon.table.source.Split;
@@ -125,7 +126,16 @@ List<PartitionEntry> listPartitionEntries() {
         List<PartitionEntry> partitionEntries = new ArrayList<>();
         for (Pair<LinkedHashMap<String, String>, Path> partition2Path : partition2Paths) {
             BinaryRow row = toPartitionRow(partition2Path.getKey());
-            partitionEntries.add(new PartitionEntry(row, -1L, -1L, -1L, -1L, -1));
+            // Discovering partitions from directories measures nothing about what is inside them,
+            // so every statistic is unknown rather than zero.
+            partitionEntries.add(
+                    new PartitionEntry(
+                            row,
+                            PartitionStatistics.UNKNOWN,
+                            PartitionStatistics.UNKNOWN,
+                            PartitionStatistics.UNKNOWN,
+                            PartitionStatistics.UNKNOWN,
+                            PartitionStatistics.UNKNOWN_TOTAL_BUCKETS));
         }
         return partitionEntries;
     }

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileWriter.java

@@ -24,7 +24,7 @@
 import org.apache.paimon.format.FileFormat;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
-import org.apache.paimon.fs.TwoPhaseOutputStream;
+import org.apache.paimon.io.FormatTableWrittenFile;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.FileStorePathFactory;
@@ -97,15 +97,15 @@ public void close() throws Exception {
     }
 
     public List<CommitMessage> prepareCommit() throws Exception {
-        List<TwoPhaseOutputStream.Committer> committers = new ArrayList<>();
+        List<FormatTableWrittenFile> writtenFiles = new ArrayList<>();
         try {
             for (FormatTableRecordWriter writer : writers.values()) {
-                committers.addAll(writer.closeAndGetCommitters());
+                writtenFiles.addAll(writer.closeAndGetWrittenFiles());
             }
         } catch (Exception e) {
-            for (TwoPhaseOutputStream.Committer committer : committers) {
+            for (FormatTableWrittenFile writtenFile : writtenFiles) {
                 try {
-                    committer.discard(fileIO);
+                    writtenFile.committer().discard(fileIO);
                 } catch (Exception cleanupException) {
                     e.addSuppressed(cleanupException);
                 }
@@ -119,8 +119,12 @@ public List<CommitMessage> prepareCommit() throws Exception {
         }
 
         List<CommitMessage> commitMessages = new ArrayList<>();
-        for (TwoPhaseOutputStream.Committer committer : committers) {
-            commitMessages.add(new TwoPhaseCommitMessage(committer));
+        for (FormatTableWrittenFile writtenFile : writtenFiles) {
+            commitMessages.add(
+                    new TwoPhaseCommitMessage(
+                            writtenFile.committer(),
+                            writtenFile.recordCount(),
+                            writtenFile.fileSizeInBytes()));
         }
         return commitMessages;
     }

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRecordWriter.java

@@ -21,9 +21,9 @@
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.format.FileFormat;
 import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.fs.TwoPhaseOutputStream;
 import org.apache.paimon.io.DataFilePathFactory;
 import org.apache.paimon.io.FormatTableRollingFileWriter;
+import org.apache.paimon.io.FormatTableWrittenFile;
 import org.apache.paimon.types.RowType;
 
 import java.util.ArrayList;
@@ -65,14 +65,14 @@ public void write(InternalRow data) throws Exception {
         writer.write(data);
     }
 
-    public List<TwoPhaseOutputStream.Committer> closeAndGetCommitters() throws Exception {
-        List<TwoPhaseOutputStream.Committer> commits = new ArrayList<>();
+    public List<FormatTableWrittenFile> closeAndGetWrittenFiles() throws Exception {
+        List<FormatTableWrittenFile> written = new ArrayList<>();
         if (writer != null) {
             writer.close();
-            commits.addAll(writer.committers());
+            written.addAll(writer.writtenFiles());
             writer = null;
         }
-        return commits;
+        return written;
     }
 
     @Override