This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b297d9079a [core] Carry the row count and byte size a format table
writer already counted (#9121)
b297d9079a is described below
commit b297d9079ad3d6197c651a367ba0889ce9bbdd25
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Mon Aug 10 13:45:04 2026 +0800
[core] Carry the row count and byte size a format table writer already
counted (#9121)
---
.../paimon/partition/PartitionStatistics.java | 42 ++++++++++++++++++++--
.../paimon/partition/PartitionStatisticsTest.java | 32 +++++++++++++++++
.../paimon/io/FormatTableRollingFileWriter.java | 23 ++++++++----
.../paimon/io/FormatTableSingleFileWriter.java | 18 ++++++++++
.../FormatTableWrittenFile.java} | 40 ++++++++++-----------
.../table/format/FileSystemSplitEnumerator.java | 12 ++++++-
.../paimon/table/format/FormatTableFileWriter.java | 18 ++++++----
.../table/format/FormatTableRecordWriter.java | 10 +++---
.../paimon/table/format/TwoPhaseCommitMessage.java | 30 +++++++++++++++-
.../paimon/table/format/FormatTableWriteTest.java | 38 ++++++++++++++------
10 files changed, 209 insertions(+), 54 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 ab87f02ed6..6717bbc258 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.Map;
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
@@ -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 class PartitionStatistics implements Serializable {
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;
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 d9fd8e8bb1..ec7f12e933 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.apache.paimon.utils.JsonSerdeUtil;
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 @@ public class PartitionStatisticsTest {
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();
+ }
}
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 96e297c327..31ed14fbb1 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<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 class FormatTableRollingFileWriter implements
AutoCloseable {
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 @@ public class FormatTableRollingFileWriter implements
AutoCloseable {
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 class FormatTableRollingFileWriter implements
AutoCloseable {
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
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 6290d95391..10e93f7989 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 class FormatTableSingleFileWriter {
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 class FormatTableSingleFileWriter {
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/table/format/TwoPhaseCommitMessage.java
b/paimon-core/src/main/java/org/apache/paimon/io/FormatTableWrittenFile.java
similarity index 54%
copy from
paimon-core/src/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java
copy to
paimon-core/src/main/java/org/apache/paimon/io/FormatTableWrittenFile.java
index ffb08064dd..f40da268e1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/FormatTableWrittenFile.java
@@ -16,39 +16,37 @@
* limitations under the License.
*/
-package org.apache.paimon.table.format;
+package org.apache.paimon.io;
-import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.fs.TwoPhaseOutputStream;
-import org.apache.paimon.table.sink.CommitMessage;
-import javax.annotation.Nullable;
-
-/** {@link CommitMessage} implementation for format table. */
-public class TwoPhaseCommitMessage implements CommitMessage {
+/**
+ * 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 TwoPhaseCommitMessage(TwoPhaseOutputStream.Committer committer) {
+ public FormatTableWrittenFile(
+ TwoPhaseOutputStream.Committer committer, long recordCount, long
fileSizeInBytes) {
this.committer = committer;
+ this.recordCount = recordCount;
+ this.fileSizeInBytes = fileSizeInBytes;
}
- @Override
- public BinaryRow partition() {
- return null;
- }
-
- @Override
- public int bucket() {
- return 0;
+ public TwoPhaseOutputStream.Committer committer() {
+ return committer;
}
- @Override
- public @Nullable Integer totalBuckets() {
- return 0;
+ public long recordCount() {
+ return recordCount;
}
- public TwoPhaseOutputStream.Committer getCommitter() {
- return committer;
+ 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 7373e35d4d..7d4bccfd15 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.fs.Path;
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 @@ final class FileSystemSplitEnumerator extends
SplitEnumerator {
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;
}
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 9f18ee5758..a3b48ee7fa 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.data.InternalRow;
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 class FormatTableFileWriter {
}
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 class FormatTableFileWriter {
}
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;
}
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 83677d9448..3e54f27df9 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 @@ package org.apache.paimon.table.format;
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 class FormatTableRecordWriter implements
AutoCloseable {
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
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 ffb08064dd..f44c5e9b89 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 @@ package org.apache.paimon.table.format;
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.
+ *
+ * <p>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 class TwoPhaseCommitMessage implements CommitMessage {
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 3864f39ad6..4ca81e6a0b 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.IOException;
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 @@ class FormatTableWriteTest {
}
assertThat(messages).hasSize(3);
- List<Path> 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<Path, Long> reportedSizes = new LinkedHashMap<>();
+ for (CommitMessage message : messages) {
+ TwoPhaseCommitMessage twoPhase = (TwoPhaseCommitMessage) message;
+ reportedSizes.put(twoPhase.getCommitter().targetPath(),
twoPhase.fileSizeInBytes());
+ }
+ assertThat(reportedSizes).hasSize(messages.size());
+ List<Path> dataFiles = new ArrayList<>(reportedSizes.keySet());
try (BatchTableCommit commit = writeBuilder.newCommit()) {
commit.commit(messages);
}
+ for (Map.Entry<Path, Long> reported : reportedSizes.entrySet()) {
+ assertThat(reported.getValue())
+ .as("byte count reported for %s", reported.getKey())
+ .isEqualTo(fileIO.getFileSize(reported.getKey()));
+ }
+
List<Long> rowCounts =
dataFiles.stream()
.map(
@@ -151,11 +167,13 @@ class FormatTableWriteTest {
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");
});
