[core][rest] Report partition statistics when registering partitions

docs/static/rest-catalog-open-api.yaml

@@ -2489,6 +2489,14 @@ components:
         ignoreIfExists:
           type: boolean
           default: true
+        partitionStatistics:
+          description: Statistics of the partitions being created, matched to partitionSpecs by their spec rather than by position, so they may cover only some of them. A field left negative was never measured, which is not the same as zero.
+          type: [ array, "null" ]
+          items:
+            $ref: '#/components/schemas/PartitionStatistics'
+        replaceStatistics:
+          description: Whether partitionStatistics replace the stored values rather than add to them; required whenever partitionStatistics is present, and absent otherwise. Replacing overwrites recordCount, fileSizeInBytes, fileCount and lastFileCreationTime; adding sums the three counts and keeps the later lastFileCreationTime, since two timestamps do not add. A field reported as unknown leaves the stored one alone either way, and totalBuckets is never combined. A client that reports only the files it just wrote adds; one that reports a whole partition, such as an overwrite or a directory rescan, replaces.
+          type: [ boolean, "null" ]
     CreatePartitionsResponse:
       type: object
       required:

paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java

@@ -872,16 +872,39 @@ public void markDonePartitions(Identifier identifier, List<Map<String, String>>
                 restAuthFunction);
     }
 
-    /** Create partitions for table, ignoring partitions which already exist. */
-    public CreatePartitionsResponse createPartitions(
-            Identifier identifier, List<Map<String, String>> partitions) {
-        return createPartitions(identifier, partitions, true);
-    }
-
-    /** Create partitions for table. */
+    /**
+     * Create partitions for table, optionally reporting their statistics in the same request, so
+     * that a partition is never registered by a request whose statistics failed on their own. A
+     * server that stores no statistics still registers the partitions.
+     *
+     * <p>How a report combines with the stored values is per field. Replacing overwrites all four
+     * of recordCount, fileSizeInBytes, fileCount and lastFileCreationTime; adding sums the three
+     * counts and keeps the later creation time, since two timestamps do not add. A field reported
+     * as unknown leaves the stored one alone either way, and a report never creates or removes a
+     * partition row.
+     *
+     * @param identifier database name and table name
+     * @param partitions partitions to be created
+     * @param ignoreIfExists if false, fail when any partition already exists and apply none of the
+     *     batch
+     * @param statistics statistics to report, matched to {@code partitions} by {@link
+     *     PartitionStatistics#spec()} rather than by position, or null to report none
+     * @param replaceStatistics whether the report replaces the stored values rather than adding to
+     *     them; ignored when {@code statistics} is null, and not sent at all in that case
+     * @return the partitions the server created and the ones it already held
+     */
     public CreatePartitionsResponse createPartitions(
-            Identifier identifier, List<Map<String, String>> partitions, boolean ignoreIfExists) {
-        CreatePartitionsRequest request = new CreatePartitionsRequest(partitions, ignoreIfExists);
+            Identifier identifier,
+            List<Map<String, String>> partitions,
+            boolean ignoreIfExists,
+            @Nullable List<PartitionStatistics> statistics,
+            boolean replaceStatistics) {
+        CreatePartitionsRequest request =
+                new CreatePartitionsRequest(
+                        partitions,
+                        ignoreIfExists,
+                        statistics,
+                        statistics == null ? null : replaceStatistics);
         return client.post(
                 resourcePaths.partitions(identifier.getDatabaseName(), identifier.getObjectName()),
                 request,

paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java

@@ -18,41 +18,72 @@
 
 package org.apache.paimon.rest.requests;
 
+import org.apache.paimon.partition.PartitionStatistics;
 import org.apache.paimon.rest.RESTRequest;
 
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 import javax.annotation.Nullable;
 
 import java.util.List;
 import java.util.Map;
 
-/** Request for creating partitions. */
+/**
+ * Request for creating partitions.
+ *
+ * <p>Statistics ride along optionally, matched to {@code partitionSpecs} by {@link
+ * PartitionStatistics#spec()} rather than by position, so they may cover only some of them. Both
+ * statistics fields are absent unless the client reports.
+ */
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class CreatePartitionsRequest implements RESTRequest {
 
     private static final String FIELD_PARTITION_SPECS = "partitionSpecs";
     private static final String FIELD_IGNORE_IF_EXISTS = "ignoreIfExists";
+    private static final String FIELD_PARTITION_STATISTICS = "partitionStatistics";
+    private static final String FIELD_REPLACE_STATISTICS = "replaceStatistics";
 
     @JsonProperty(FIELD_PARTITION_SPECS)
     private final List<Map<String, String>> partitionSpecs;
 
     @JsonProperty(FIELD_IGNORE_IF_EXISTS)
     private final boolean ignoreIfExists;
 
+    @JsonProperty(FIELD_PARTITION_STATISTICS)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Nullable
+    private final List<PartitionStatistics> partitionStatistics;
+
+    @JsonProperty(FIELD_REPLACE_STATISTICS)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Nullable
+    private final Boolean replaceStatistics;
+
     public CreatePartitionsRequest(List<Map<String, String>> partitionSpecs) {
         this(partitionSpecs, true);
     }
 
+    public CreatePartitionsRequest(
+            List<Map<String, String>> partitionSpecs, @Nullable Boolean ignoreIfExists) {
+        this(partitionSpecs, ignoreIfExists, null, null);
+    }
+
     @JsonCreator
     public CreatePartitionsRequest(
             @JsonProperty(FIELD_PARTITION_SPECS) List<Map<String, String>> partitionSpecs,
-            @JsonProperty(FIELD_IGNORE_IF_EXISTS) @Nullable Boolean ignoreIfExists) {
+            @JsonProperty(FIELD_IGNORE_IF_EXISTS) @Nullable Boolean ignoreIfExists,
+            @JsonProperty(FIELD_PARTITION_STATISTICS) @Nullable
+                    List<PartitionStatistics> partitionStatistics,
+            @JsonProperty(FIELD_REPLACE_STATISTICS) @Nullable Boolean replaceStatistics) {
         this.partitionSpecs = partitionSpecs;
         this.ignoreIfExists = ignoreIfExists == null || ignoreIfExists;
+        this.partitionStatistics = partitionStatistics;
+        this.replaceStatistics = replaceStatistics;
     }
 
     @JsonGetter(FIELD_PARTITION_SPECS)
@@ -64,4 +95,35 @@ public List<Map<String, String>> getPartitionSpecs() {
     public boolean ignoreIfExists() {
         return ignoreIfExists;
     }
+
+    /** Reported statistics, or null when the client reports none. */
+    @JsonGetter(FIELD_PARTITION_STATISTICS)
+    @Nullable
+    public List<PartitionStatistics> getPartitionStatistics() {
+        return partitionStatistics;
+    }
+
+    /**
+     * Whether the reported statistics replace what the catalog holds rather than adding to it; null
+     * when none are reported.
+     */
+    @JsonGetter(FIELD_REPLACE_STATISTICS)
+    @Nullable
+    public Boolean replaceStatistics() {
+        return replaceStatistics;
+    }
+
+    /**
+     * Registering is an upsert and replacing lands on the same value twice, so both survive being
+     * sent again. Adding does not: a second delivery is counted again. A request that reports no
+     * statistics increments nothing and so keeps its retry, which is the shape batching a create
+     * leaves behind.
+     */
+    @JsonIgnore
+    @Override
+    public boolean isRetrySafe() {
+        return partitionStatistics == null
+                || partitionStatistics.isEmpty()
+                || Boolean.TRUE.equals(replaceStatistics);
+    }
 }

paimon-api/src/test/java/org/apache/paimon/rest/HttpClientRetrySafetyTest.java

@@ -18,8 +18,10 @@
 
 package org.apache.paimon.rest;
 
+import org.apache.paimon.partition.PartitionStatistics;
 import org.apache.paimon.rest.exceptions.ServiceUnavailableException;
-import org.apache.paimon.rest.responses.ListDatabasesResponse;
+import org.apache.paimon.rest.requests.CreatePartitionsRequest;
+import org.apache.paimon.rest.responses.CreatePartitionsResponse;
 
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
 
@@ -33,21 +35,26 @@
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
 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.
+ * Tests that a POST the server cannot absorb twice is sent exactly once, and that every other POST
+ * keeps the 429/503 retry it has always had.
  *
  * <p>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 static final String PATH = "/partitions";
+
+    private static final Map<String, String> SPEC = Collections.singletonMap("dt", "20260728");
 
     private HttpServer server;
     private HttpClient client;
@@ -61,10 +68,10 @@ public void setUp() throws Exception {
                 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.
+                        // request; this is exactly the shape that double counts an ADD report.
                         respond(exchange, 503, "{\"message\":\"busy\",\"code\":503}");
                     } else {
-                        respond(exchange, 200, "{\"databases\":[\"db\"]}");
+                        respond(exchange, 200, "{\"created\":[],\"existed\":[]}");
                     }
                 });
         server.start();
@@ -79,20 +86,34 @@ public void tearDown() {
     }
 
     @Test
-    public void testARequestThatDeclaresItselfUnsafeIsNotRetried() {
-        assertThatThrownBy(() -> post(new UnsafeToRetry()))
+    public void testAnAddingReportIsNotRetried() {
+        assertThatThrownBy(() -> post(request(false, statistics())))
                 .isInstanceOf(ServiceUnavailableException.class);
 
-        // Retrying would apply the same request a second time, and nothing downstream could see it.
+        // Retrying would add the same increment a second time.
         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.
+    public void testAReplacingReportIsRetried() {
+        assertThat(post(request(true, statistics()))).isNotNull();
+
+        // Replacing lands on the same value, so a second delivery changes nothing.
+        assertThat(requests.get()).isEqualTo(2);
+    }
+
+    @Test
+    public void testARequestCarryingNoReportIsRetried() {
+        assertThat(post(new CreatePartitionsRequest(Collections.singletonList(SPEC)))).isNotNull();
+
+        assertThat(requests.get()).isEqualTo(2);
+    }
+
+    @Test
+    public void testARequestThatNeverHeardOfReportsKeepsItsRetry() {
+        // Nearly every request type rides the interface default, so every case above would still
+        // pass if that default flipped to false and took 429/503 retry away from every other POST
+        // in the catalog. This is the only case that rides it.
         assertThat(new DefaultRetrySafety().isRetrySafe()).isTrue();
         assertThat(post(new DefaultRetrySafety())).isNotNull();
 
@@ -101,37 +122,57 @@ public void testARequestThatNeverHeardOfRetrySafetyKeepsItsRetry() {
 
     @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");
+        // isRetrySafe is how the client treats the request, not something the server is told.
+        // Without the @JsonIgnore on the interface default it would show up in every body.
         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 {
+    @Test
+    public void testAReportOfNoStatisticsIsRetried() {
+        // Splitting a large create leaves batches with an empty statistics list; they increment
+        // nothing, so they keep their retry.
+        assertThat(post(request(false, Collections.emptyList()))).isNotNull();
 
-        @JsonGetter("name")
-        public String getName() {
-            return "db";
-        }
+        assertThat(requests.get()).isEqualTo(2);
     }
 
-    /** A request that must reach the server at most once. */
-    private static class UnsafeToRetry implements RESTRequest {
+    @Test
+    public void testOnlyANonEmptyAddingReportDeclaresItselfUnsafeToRetry() {
+        assertThat(request(false, statistics()).isRetrySafe()).isFalse();
+        assertThat(request(false, Collections.emptyList()).isRetrySafe()).isTrue();
+        assertThat(request(true, statistics()).isRetrySafe()).isTrue();
+        assertThat(new CreatePartitionsRequest(Collections.singletonList(SPEC)).isRetrySafe())
+                .isTrue();
+        // A request that leaves the flag out reports nothing, so it is replayable whatever the
+        // flag would have said.
+        assertThat(
+                        new CreatePartitionsRequest(
+                                        Collections.singletonList(SPEC), true, null, null)
+                                .isRetrySafe())
+                .isTrue();
+    }
 
-        @JsonGetter("name")
-        public String getName() {
-            return "db";
-        }
+    /** A request that leaves {@link RESTRequest#isRetrySafe()} at its default, as nearly all do. */
+    private static class DefaultRetrySafety implements RESTRequest {
 
-        @Override
-        public boolean isRetrySafe() {
-            return false;
+        @JsonGetter("partitionSpecs")
+        public List<Map<String, String>> getPartitionSpecs() {
+            return Collections.singletonList(SPEC);
         }
     }
 
-    private ListDatabasesResponse post(RESTRequest request) {
-        return client.post(PATH, request, ListDatabasesResponse.class, null);
+    private CreatePartitionsResponse post(RESTRequest request) {
+        return client.post(PATH, request, CreatePartitionsResponse.class, null);
+    }
+
+    private static CreatePartitionsRequest request(
+            boolean replaceStatistics, List<PartitionStatistics> statistics) {
+        return new CreatePartitionsRequest(
+                Collections.singletonList(SPEC), true, statistics, replaceStatistics);
+    }
+
+    private static List<PartitionStatistics> statistics() {
+        return Collections.singletonList(new PartitionStatistics(SPEC, 3L, 300L, 1L, 1000L, -1));
     }
 
     private static void respond(HttpExchange exchange, int statusCode, String body)

paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java

@@ -346,9 +346,14 @@ public void createPartitions(Identifier identifier, List<Map<String, String>> pa
 
     @Override
     public void createPartitions(
-            Identifier identifier, List<Map<String, String>> partitions, boolean ignoreIfExists)
+            Identifier identifier,
+            List<Map<String, String>> partitions,
+            boolean ignoreIfExists,
+            @Nullable List<PartitionStatistics> statistics,
+            boolean replaceStatistics)
             throws TableNotExistException {
-        wrapped.createPartitions(identifier, partitions, ignoreIfExists);
+        wrapped.createPartitions(
+                identifier, partitions, ignoreIfExists, statistics, replaceStatistics);
         if (partitionCache != null) {
             partitionCache.invalidate(identifier);
         }