From 73f75b0e715729a6d7e8685a3ac5d2e8f5c6a499 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Sun, 9 Aug 2026 01:47:34 +0800 Subject: [PATCH 1/3] [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/3] [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"); }); From b467d1b30ecf22c9d2eeead1268e0e447fe495a3 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Sun, 9 Aug 2026 01:50:35 +0800 Subject: [PATCH 3/3] [rest] Do not replay a POST the server cannot absorb twice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The REST client retries a 429 or a 503 on any request, POST included. That is right for nearly everything it sends: registering a partition, creating a database, committing a snapshot the server already holds all land on the same state the second time, and the retry is the only defence against a rate limiter or a restarting node. It is wrong for a request that reports an increment. A 429 or a 503 can reach the client from an intermediary after the server already applied the request, so replaying it applies it again — and an increment applied twice is a wrong number that no caller can see. Nothing in the response distinguishes the two cases, which is why this has to be decided by the request rather than by the status. RESTRequest gains isRetrySafe(), defaulting to true so every existing request keeps the retry it has today. A request answering false is sent exactly once and the failure reaches the caller, which knows whether re-sending is safe. The mark travels in the HttpClientContext rather than in the request, so it never reaches the wire and survives whatever the exec chain does to the request object; @JsonIgnore keeps the getter out of the serialized body as well, and a test pins both. --- .../ExponentialHttpRequestRetryStrategy.java | 29 ++++ .../org/apache/paimon/rest/HttpClient.java | 25 ++- .../org/apache/paimon/rest/RESTRequest.java | 24 ++- .../rest/HttpClientRetrySafetyTest.java | 146 ++++++++++++++++++ ...stExponentialHttpRequestRetryStrategy.java | 18 +++ 5 files changed, 237 insertions(+), 5 deletions(-) create mode 100644 paimon-api/src/test/java/org/apache/paimon/rest/HttpClientRetrySafetyTest.java diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/ExponentialHttpRequestRetryStrategy.java b/paimon-api/src/main/java/org/apache/paimon/rest/ExponentialHttpRequestRetryStrategy.java index e8e621a64dfc..08e7215c11a2 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/ExponentialHttpRequestRetryStrategy.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/ExponentialHttpRequestRetryStrategy.java @@ -23,6 +23,7 @@ import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableSet; import org.apache.hc.client5.http.HttpRequestRetryStrategy; +import org.apache.hc.client5.http.protocol.HttpClientContext; import org.apache.hc.client5.http.utils.DateUtils; import org.apache.hc.core5.concurrent.CancellableDependency; import org.apache.hc.core5.http.ConnectionClosedException; @@ -35,6 +36,7 @@ import org.apache.hc.core5.http.protocol.HttpContext; import org.apache.hc.core5.util.TimeValue; +import javax.annotation.Nullable; import javax.net.ssl.SSLException; import java.io.IOException; @@ -47,6 +49,16 @@ import java.util.concurrent.ThreadLocalRandom; class ExponentialHttpRequestRetryStrategy implements HttpRequestRetryStrategy { + + /** + * Context attribute marking one exchange as "must not be sent twice". A 429 or a 503 can reach + * the client from a proxy after the server already applied the request, so replaying it applies + * it again; for a request that is not idempotent by content that is a silent double apply. The + * mark travels in the context rather than in the request, so it never reaches the wire and + * survives whatever the exec chain does to the request object. + */ + static final String RETRY_UNSAFE_ATTRIBUTE = "paimon.rest.retry-unsafe"; + private final int maxRetries; private final Set> nonRetriableExceptions; private final Set retriableCodes; @@ -98,9 +110,26 @@ public boolean retryRequest( @Override public boolean retryRequest(HttpResponse response, int execCount, HttpContext context) { + if (isRetryUnsafe(context)) { + // The status says nothing about whether the server applied the request: a 503 from an + // intermediary can follow a request that already took effect. Replaying it would apply + // it twice with nobody the wiser, so the failure goes back to the caller instead. + return false; + } return execCount <= maxRetries && retriableCodes.contains(response.getCode()); } + /** A context for one exchange that must be sent exactly once. */ + static HttpClientContext retryUnsafeContext() { + HttpClientContext context = HttpClientContext.create(); + context.setAttribute(RETRY_UNSAFE_ATTRIBUTE, Boolean.TRUE); + return context; + } + + static boolean isRetryUnsafe(@Nullable HttpContext context) { + return context != null && Boolean.TRUE.equals(context.getAttribute(RETRY_UNSAFE_ATTRIBUTE)); + } + @Override public TimeValue getRetryInterval(HttpResponse response, int execCount, HttpContext context) { // a server may send a 429 / 503 with a Retry-After header diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java index 311a1a2d4abe..8205dfe21295 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java @@ -36,8 +36,12 @@ import org.apache.hc.core5.http.ClassicHttpResponse; import org.apache.hc.core5.http.ContentType; import org.apache.hc.core5.http.Header; +import org.apache.hc.core5.http.io.HttpClientResponseHandler; import org.apache.hc.core5.http.io.entity.StringEntity; import org.apache.hc.core5.http.message.BasicHeader; +import org.apache.hc.core5.http.protocol.HttpContext; + +import javax.annotation.Nullable; import java.io.IOException; import java.util.Arrays; @@ -100,7 +104,13 @@ public T post( } Header[] authHeaders = getHeaders(path, "POST", encodedBody, restAuthFunction); httpPost.setHeaders(authHeaders); - return exec(httpPost, responseType); + // A POST the server cannot absorb twice is sent exactly once, whatever the status says. + return exec( + httpPost, + responseType, + body != null && !body.isRetrySafe() + ? ExponentialHttpRequestRetryStrategy.retryUnsafeContext() + : null); } @Override @@ -127,9 +137,13 @@ void setErrorHandler(ErrorHandler errorHandler) { } private T exec(HttpUriRequestBase request, Class responseType) { + return exec(request, responseType, null); + } + + private T exec( + HttpUriRequestBase request, Class responseType, @Nullable HttpContext context) { try { - return DEFAULT_HTTP_CLIENT.execute( - request, + HttpClientResponseHandler handler = response -> { String responseBodyStr = RESTUtil.extractResponseBodyAsString(response); if (!RESTUtil.isSuccessful(response)) { @@ -159,7 +173,10 @@ private T exec(HttpUriRequestBase request, Class res } else { throw new RESTException("response body is null."); } - }); + }; + return context == null + ? DEFAULT_HTTP_CLIENT.execute(request, handler) + : DEFAULT_HTTP_CLIENT.execute(request, context, handler); } catch (IOException e) { // No cause: a redirect/protocol error message can echo the target URL (a signed URL). throw new RESTException( diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTRequest.java b/paimon-api/src/main/java/org/apache/paimon/rest/RESTRequest.java index 9c6758df14f0..35412bcb2230 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTRequest.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTRequest.java @@ -18,5 +18,27 @@ package org.apache.paimon.rest; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore; + /** Interface to mark a REST request. */ -public interface RESTRequest extends RESTMessage {} +public interface RESTRequest extends RESTMessage { + + /** + * Whether sending this request a second time leaves the server where sending it once does. + * + *

This is how the client treats the request, not something the server is told: it is a + * getter on a serialized type and must stay off the wire. + * + *

POST is not idempotent by method, but nearly every request Paimon sends over it is by + * content — registering a partition, creating a database, committing a snapshot the server + * already holds — so the client retries them after a 429 or a 503, which is the only defence + * against a rate limiter or a restarting node. A request that reports an increment is the + * exception: a proxy answering 503 after the server already applied it turns an automatic retry + * into a double count that no caller can see. Such a request says so here and is sent exactly + * once; the failure reaches the caller, which can decide. + */ + @JsonIgnore + default boolean isRetrySafe() { + return true; + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientRetrySafetyTest.java b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientRetrySafetyTest.java new file mode 100644 index 000000000000..8aca4d994776 --- /dev/null +++ b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientRetrySafetyTest.java @@ -0,0 +1,146 @@ +/* + * 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.rest; + +import org.apache.paimon.rest.exceptions.ServiceUnavailableException; +import org.apache.paimon.rest.responses.ListDatabasesResponse; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests that a POST declaring itself unsafe to replay is sent exactly once, and that every other + * POST keeps the 429/503 retry it has always had. + * + *

The server here refuses only the first attempt, so a retried request succeeds on its second + * one: the request count separates "sent once" from "sent again" without waiting out five backoffs. + */ +public class HttpClientRetrySafetyTest { + + private static final String PATH = "/databases"; + + private HttpServer server; + private HttpClient client; + private final AtomicInteger requests = new AtomicInteger(); + + @BeforeEach + public void setUp() throws Exception { + server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + PATH, + exchange -> { + if (requests.incrementAndGet() == 1) { + // A proxy answering 503 says nothing about whether the server applied the + // request; this is exactly the shape that applies a request twice. + respond(exchange, 503, "{\"message\":\"busy\",\"code\":503}"); + } else { + respond(exchange, 200, "{\"databases\":[\"db\"]}"); + } + }); + server.start(); + client = new HttpClient("http://127.0.0.1:" + server.getAddress().getPort()); + } + + @AfterEach + public void tearDown() { + if (server != null) { + server.stop(0); + } + } + + @Test + public void testARequestThatDeclaresItselfUnsafeIsNotRetried() { + assertThatThrownBy(() -> post(new UnsafeToRetry())) + .isInstanceOf(ServiceUnavailableException.class); + + // Retrying would apply the same request a second time, and nothing downstream could see it. + assertThat(requests.get()).isEqualTo(1); + } + + @Test + public void testARequestThatNeverHeardOfRetrySafetyKeepsItsRetry() { + // The regression this guards against is the global one: every request type implements + // RESTRequest and rides the interface default, so the case above would still pass if the + // default flipped to false and silently took 429/503 retry away from commits, database + // creation and every other POST in the catalog. + assertThat(new DefaultRetrySafety().isRetrySafe()).isTrue(); + assertThat(post(new DefaultRetrySafety())).isNotNull(); + + assertThat(requests.get()).isEqualTo(2); + } + + @Test + public void testRetrySafetyNeverReachesTheWire() { + // isRetrySafe is how the client treats the request, not something the server is told. It is + // a getter on a serialized type, so without @JsonIgnore it would show up in the body. + assertThat(RESTUtil.encodedBody(new UnsafeToRetry())).doesNotContain("retrySafe"); + assertThat(RESTUtil.encodedBody(new DefaultRetrySafety())).doesNotContain("retrySafe"); + } + + /** A request that leaves {@link RESTRequest#isRetrySafe()} at its default, as all others do. */ + private static class DefaultRetrySafety implements RESTRequest { + + @JsonGetter("name") + public String getName() { + return "db"; + } + } + + /** A request that must reach the server at most once. */ + private static class UnsafeToRetry implements RESTRequest { + + @JsonGetter("name") + public String getName() { + return "db"; + } + + @Override + public boolean isRetrySafe() { + return false; + } + } + + private ListDatabasesResponse post(RESTRequest request) { + return client.post(PATH, request, ListDatabasesResponse.class, null); + } + + private static void respond(HttpExchange exchange, int statusCode, String body) + throws IOException { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(statusCode, bytes.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(bytes); + } + } +} diff --git a/paimon-api/src/test/java/org/apache/paimon/rest/TestExponentialHttpRequestRetryStrategy.java b/paimon-api/src/test/java/org/apache/paimon/rest/TestExponentialHttpRequestRetryStrategy.java index 9a743a93d913..cc4863f4e80d 100644 --- a/paimon-api/src/test/java/org/apache/paimon/rest/TestExponentialHttpRequestRetryStrategy.java +++ b/paimon-api/src/test/java/org/apache/paimon/rest/TestExponentialHttpRequestRetryStrategy.java @@ -20,6 +20,7 @@ import org.apache.hc.client5.http.HttpRequestRetryStrategy; import org.apache.hc.client5.http.classic.methods.HttpGet; +import org.apache.hc.client5.http.protocol.HttpClientContext; import org.apache.hc.client5.http.utils.DateUtils; import org.apache.hc.core5.http.ConnectionClosedException; import org.apache.hc.core5.http.HttpHeaders; @@ -220,4 +221,21 @@ public void testRetryDoesNotHappenOnUnacceptableStatusCodes(int statusCode) { BasicHttpResponse response = new BasicHttpResponse(statusCode, String.valueOf(statusCode)); assertThat(retryStrategy.retryRequest(response, 3, null)).isFalse(); } + + @ParameterizedTest + @ValueSource(ints = {429, 503}) + public void testRetryDoesNotHappenOnAnExchangeMarkedUnsafe(int statusCode) { + BasicHttpResponse response = new BasicHttpResponse(statusCode, String.valueOf(statusCode)); + + // The status says the request failed, not that the server never saw it: a 503 can come + // from an intermediary after the request already took effect. + assertThat( + retryStrategy.retryRequest( + response, + 1, + ExponentialHttpRequestRetryStrategy.retryUnsafeContext())) + .isFalse(); + // Only the marked exchange loses the retry; a plain one keeps it. + assertThat(retryStrategy.retryRequest(response, 1, HttpClientContext.create())).isTrue(); + } }