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/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/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; } 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"); });