From 73f75b0e715729a6d7e8685a3ac5d2e8f5c6a499 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Sun, 9 Aug 2026 01:47:34 +0800 Subject: [PATCH 1/2] [core] Spell out what a negative partition statistic means MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PartitionStatistics said only that its fields "may be negative, indicating that some data has been removed". That covers one of the two planes the class is read on, and consumers have been getting the other one wrong. On the delta plane — what a commit changed — a negative value is a decrement the server adds to what it holds. That is the existing meaning and nothing here changes it. On the observation plane — what listPartitions returns for a partition as it stands — a negative value means nobody ever reported that field, and 0 means an exact zero. Conflating them is not cosmetic: a consumer that reads unknown as zero plans against an empty partition that may hold a billion rows, and one that does arithmetic on it gets a number that is wrong rather than missing. So the plane is named in the javadoc, unknown gets a name (UNKNOWN, with isKnown() to test it rather than each caller comparing against -1), and unknown is documented as per field: a reporter that only knows the file count leaves the record count unknown and fills the rest. The fields stay primitive. Boxing them to express unknown as null would be a breaking change to a @Public class, and the encoding above needs no new type. FileSystemSplitEnumerator now says PartitionStatistics.UNKNOWN where it said -1. Discovering partitions by listing directories measures nothing about what is inside them, which is what unknown already meant there; this is the same value under its own name. --- .../paimon/partition/PartitionStatistics.java | 42 ++++++++++++++++++- .../partition/PartitionStatisticsTest.java | 32 ++++++++++++++ .../format/FileSystemSplitEnumerator.java | 12 +++++- 3 files changed, 83 insertions(+), 3 deletions(-) diff --git a/paimon-api/src/main/java/org/apache/paimon/partition/PartitionStatistics.java b/paimon-api/src/main/java/org/apache/paimon/partition/PartitionStatistics.java index ab87f02ed6a1..6717bbc258ea 100644 --- a/paimon-api/src/main/java/org/apache/paimon/partition/PartitionStatistics.java +++ b/paimon-api/src/main/java/org/apache/paimon/partition/PartitionStatistics.java @@ -30,8 +30,23 @@ import java.util.Objects; /** - * Statistics of a partition, fields inside may be negative, indicating that some data has been - * removed. + * Statistics of a partition. + * + *

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: + * + *

+ * + *

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 @@ -39,6 +54,15 @@ 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 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 spec() { return spec; diff --git a/paimon-api/src/test/java/org/apache/paimon/partition/PartitionStatisticsTest.java b/paimon-api/src/test/java/org/apache/paimon/partition/PartitionStatisticsTest.java index d9fd8e8bb162..ec7f12e933a6 100644 --- a/paimon-api/src/test/java/org/apache/paimon/partition/PartitionStatisticsTest.java +++ b/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 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(); + } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemSplitEnumerator.java index 7373e35d4d4c..7d4bccfd15f8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FileSystemSplitEnumerator.java +++ b/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 listPartitionEntries() { List partitionEntries = new ArrayList<>(); for (Pair, 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; } From 5ee1222dd1e03e5fdff780380721069b7a5f0721 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Sun, 9 Aug 2026 01:47:56 +0800 Subject: [PATCH 2/2] [core] Carry the row count and byte size a format table writer already counted FormatTableRollingFileWriter counts every row it writes and FormatTableSingleFileWriter knows the byte length of the file it closed. Both numbers are then dropped: closeAndGetCommitters returns only the committers, and prepareCommit wraps each one in a TwoPhaseCommitMessage that carries nothing else. Anything downstream that wants to know what a commit wrote has to go back to the filesystem and list it. This keeps the two numbers attached to the file they describe, in a new FormatTableWrittenFile that pairs the committer with them, and lets TwoPhaseCommitMessage carry it. Nothing reads them yet. TwoPhaseOutputStream.Committer is untouched. RenamingTwoPhaseOutputStream is @Public, so adding a method to the type its committer() returns would break external implementations; the counts ride the paimon-core commit message instead. --- .../io/FormatTableRollingFileWriter.java | 23 +++++--- .../io/FormatTableSingleFileWriter.java | 18 +++++++ .../paimon/io/FormatTableWrittenFile.java | 52 +++++++++++++++++++ .../table/format/FormatTableFileWriter.java | 18 ++++--- .../table/format/FormatTableRecordWriter.java | 10 ++-- .../table/format/TwoPhaseCommitMessage.java | 30 ++++++++++- .../table/format/FormatTableWriteTest.java | 38 ++++++++++---- 7 files changed, 159 insertions(+), 30 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/io/FormatTableWrittenFile.java diff --git a/paimon-core/src/main/java/org/apache/paimon/io/FormatTableRollingFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/FormatTableRollingFileWriter.java index 96e297c327f9..31ed14fbb1ae 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/FormatTableRollingFileWriter.java +++ b/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 closedWriters; - private final List committers; + private final List 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 committers() { - return committers; + /** The files this writer produced, each with the rows and bytes it holds. */ + public List writtenFiles() { + return writtenFiles; } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/io/FormatTableSingleFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/FormatTableSingleFileWriter.java index 6290d953919d..10e93f798954 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/FormatTableSingleFileWriter.java +++ b/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 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!"); diff --git a/paimon-core/src/main/java/org/apache/paimon/io/FormatTableWrittenFile.java b/paimon-core/src/main/java/org/apache/paimon/io/FormatTableWrittenFile.java new file mode 100644 index 000000000000..f40da268e1a0 --- /dev/null +++ b/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; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileWriter.java index 9f18ee5758a4..a3b48ee7fa21 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileWriter.java +++ b/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 prepareCommit() throws Exception { - List committers = new ArrayList<>(); + List 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 prepareCommit() throws Exception { } List 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; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRecordWriter.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRecordWriter.java index 83677d9448c2..3e54f27df9d0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableRecordWriter.java +++ b/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 closeAndGetCommitters() throws Exception { - List commits = new ArrayList<>(); + public List closeAndGetWrittenFiles() throws Exception { + List written = new ArrayList<>(); if (writer != null) { writer.close(); - commits.addAll(writer.committers()); + written.addAll(writer.writtenFiles()); writer = null; } - return commits; + return written; } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java b/paimon-core/src/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java index ffb08064dd17..f44c5e9b8904 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java @@ -20,17 +20,35 @@ import org.apache.paimon.data.BinaryRow; import org.apache.paimon.fs.TwoPhaseOutputStream; +import org.apache.paimon.partition.PartitionStatistics; import org.apache.paimon.table.sink.CommitMessage; import javax.annotation.Nullable; -/** {@link CommitMessage} implementation for format table. */ +/** + * {@link CommitMessage} implementation for format table. + * + *

Carries the row count and byte size of the one file it commits, counted while writing. The + * partition is not carried: {@link FormatTableCommit} derives it from the committer's target path, + * and deriving it once keeps the statistics and the registered partition from ever disagreeing. + */ public class TwoPhaseCommitMessage implements CommitMessage { + private static final long serialVersionUID = 1L; + private final TwoPhaseOutputStream.Committer committer; + private final long recordCount; + private final long fileSizeInBytes; public TwoPhaseCommitMessage(TwoPhaseOutputStream.Committer committer) { + this(committer, PartitionStatistics.UNKNOWN, PartitionStatistics.UNKNOWN); + } + + public TwoPhaseCommitMessage( + TwoPhaseOutputStream.Committer committer, long recordCount, long fileSizeInBytes) { this.committer = committer; + this.recordCount = recordCount; + this.fileSizeInBytes = fileSizeInBytes; } @Override @@ -51,4 +69,14 @@ public int bucket() { public TwoPhaseOutputStream.Committer getCommitter() { return committer; } + + /** Rows in this file, or {@link PartitionStatistics#UNKNOWN} when nobody counted them. */ + public long recordCount() { + return recordCount; + } + + /** Bytes in this file, or {@link PartitionStatistics#UNKNOWN} when nobody measured them. */ + public long fileSizeInBytes() { + return fileSizeInBytes; + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableWriteTest.java index 3864f39ad6cf..4ca81e6a0bc8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableWriteTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableWriteTest.java @@ -41,8 +41,10 @@ import java.io.InputStreamReader; import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.stream.Collectors; @@ -80,18 +82,32 @@ void testRollsByTargetRowNumber() throws Exception { } assertThat(messages).hasSize(3); - List dataFiles = - messages.stream() - .map( - message -> - ((TwoPhaseCommitMessage) message) - .getCommitter() - .targetPath()) - .collect(Collectors.toList()); + // The rows and bytes of each rolled file are counted while writing and survive to commit, + // so they never have to be recovered by reading the file back. + assertThat( + messages.stream() + .map(message -> ((TwoPhaseCommitMessage) message).recordCount()) + .collect(Collectors.toList())) + .containsExactlyInAnyOrder(2L, 2L, 1L); + // Each message carries the size of the one file it commits, not of some other file that + // happens to be positive too. + Map reportedSizes = new LinkedHashMap<>(); + for (CommitMessage message : messages) { + TwoPhaseCommitMessage twoPhase = (TwoPhaseCommitMessage) message; + reportedSizes.put(twoPhase.getCommitter().targetPath(), twoPhase.fileSizeInBytes()); + } + assertThat(reportedSizes).hasSize(messages.size()); + List dataFiles = new ArrayList<>(reportedSizes.keySet()); try (BatchTableCommit commit = writeBuilder.newCommit()) { commit.commit(messages); } + for (Map.Entry reported : reportedSizes.entrySet()) { + assertThat(reported.getValue()) + .as("byte count reported for %s", reported.getKey()) + .isEqualTo(fileIO.getFileSize(reported.getKey())); + } + List rowCounts = dataFiles.stream() .map( @@ -151,11 +167,13 @@ void testPrepareCommitFailureDiscardsPreparedFiles() throws Exception { TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); java.util.concurrent.atomic.AtomicInteger closeCount = new java.util.concurrent.atomic.AtomicInteger(); - when(recordWriter.closeAndGetCommitters()) + when(recordWriter.closeAndGetWrittenFiles()) .thenAnswer( ignored -> { if (closeCount.getAndIncrement() == 0) { - return Collections.singletonList(committer); + return Collections.singletonList( + new org.apache.paimon.io.FormatTableWrittenFile( + committer, 1L, 1L)); } throw new IOException("expected close failure"); });