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