This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a8abfafbba [core][rest] Report partition statistics when registering
partitions (#9295)
a8abfafbba is described below
commit a8abfafbbadeafad5723a51a7c888285e5d6208a
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Thu Aug 20 09:47:08 2026 +0800
[core][rest] Report partition statistics when registering partitions (#9295)
---
docs/static/rest-catalog-open-api.yaml | 8 +
.../main/java/org/apache/paimon/rest/RESTApi.java | 41 ++++-
.../rest/requests/CreatePartitionsRequest.java | 66 ++++++-
.../paimon/rest/HttpClientRetrySafetyTest.java | 109 +++++++----
.../org/apache/paimon/catalog/CachingCatalog.java | 9 +-
.../java/org/apache/paimon/catalog/Catalog.java | 41 ++++-
.../org/apache/paimon/catalog/DelegateCatalog.java | 9 +-
.../java/org/apache/paimon/rest/RESTCatalog.java | 11 +-
.../format/CatalogFormatTablePartitionManager.java | 103 ++++++++++-
.../paimon/table/format/FormatTableCommit.java | 4 +-
.../table/format/FormatTablePartitionManager.java | 28 ++-
.../apache/paimon/catalog/CachingCatalogTest.java | 26 ++-
.../apache/paimon/catalog/DelegateCatalogTest.java | 87 +++++++++
.../apache/paimon/rest/MockRESTCatalogTest.java | 160 ++++++++++++++++
.../org/apache/paimon/rest/RESTApiJsonTest.java | 49 +++++
.../org/apache/paimon/rest/RESTCatalogServer.java | 114 +++++++++++-
.../org/apache/paimon/rest/RESTCatalogTest.java | 5 +-
.../CatalogFormatTablePartitionManagerTest.java | 202 ++++++++++++++++++++-
.../format/CatalogManagedPartitionScanTest.java | 6 +-
.../format/FormatTablePartitionRepairTest.java | 7 +-
.../FormatTablePartitionDdlPlanningTest.scala | 41 +++--
.../FormatTablePartitionManagementTest.scala | 50 +++--
.../CatalogManagedPartitionMsckRepairTest.scala | 13 +-
23 files changed, 1070 insertions(+), 119 deletions(-)
diff --git a/docs/static/rest-catalog-open-api.yaml
b/docs/static/rest-catalog-open-api.yaml
index 452919f53f..8550851f50 100644
--- a/docs/static/rest-catalog-open-api.yaml
+++ b/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 o [...]
+ type: [ boolean, "null" ]
CreatePartitionsResponse:
type: object
required:
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
index 401921c8d1..238a3969ff 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -872,16 +872,39 @@ public class RESTApi {
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,
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java
index 506c66406e..dadf1f9faa 100644
---
a/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java
@@ -18,11 +18,14 @@
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;
@@ -30,12 +33,20 @@ 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;
@@ -43,16 +54,36 @@ public class CreatePartitionsRequest implements RESTRequest
{
@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 class CreatePartitionsRequest implements RESTRequest {
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);
+ }
}
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
index 8aca4d9947..d6c836dfcc 100644
---
a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientRetrySafetyTest.java
+++
b/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.IOException;
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 class HttpClientRetrySafetyTest {
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 class HttpClientRetrySafetyTest {
}
@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 class HttpClientRetrySafetyTest {
@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)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java
index 01f04cbba3..e2a2558dc0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java
@@ -346,9 +346,14 @@ public class CachingCatalog extends DelegateCatalog {
@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);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
index 5b437efbab..9653c67d01 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -31,6 +31,7 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.rest.responses.GetTagResponse;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.CatalogEnvironment;
import org.apache.paimon.table.Instant;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableSnapshot;
@@ -1040,21 +1041,23 @@ public interface Catalog extends AutoCloseable {
// ==================== Partition Modifications ==========================
/**
- * Whether this catalog supports partition modification for tables.
+ * Whether committing a table through this catalog maintains that table's
partitions in the
+ * catalog.
*
- * <p>If not, following methods will do nothing:
+ * <p>What this gates is the handler a table is given: {@link
+ * CatalogEnvironment#partitionModification()} is null for a catalog that
says false, so the
+ * commits of a table loaded from it register, alter and drop nothing. It
does not disable the
+ * methods themselves, which is how a Format Table with catalog-managed
partitions registers
+ * what a commit wrote even though its catalog reports false here:
*
* <ul>
* <li>{@link #createPartitions(Identifier, List)}.
* <li>{@link #alterPartitions(Identifier, List)}.
- * </ul>
- *
- * <p>If not, following method will be exactly the same as directly using
{@link
- * BatchTableCommit#truncatePartitions}:
- *
- * <ul>
* <li>{@link #dropPartitions(Identifier, List)}.
* </ul>
+ *
+ * <p>A catalog that keeps no partitions of its own inherits defaults that
match: creating and
+ * altering do nothing, and dropping is exactly {@link
BatchTableCommit#truncatePartitions}.
*/
boolean supportsPartitionModification();
@@ -1069,16 +1072,34 @@ public interface Catalog extends AutoCloseable {
throws TableNotExistException {}
/**
- * Create partitions of the specify table with explicit existence
semantics.
+ * Create partitions of the specify table, with explicit existence
semantics and optionally
+ * reporting statistics for them in the same call.
+ *
+ * <p>The statistics are matched to {@code partitions} by {@link
PartitionStatistics#spec()}, so
+ * they may cover only some of them, and {@code replaceStatistics} says
whether they replace
+ * what the catalog already holds or add to it. What decides whether they
survive is whether a
+ * catalog overrides this method: one that does not registers the
partitions exactly as {@link
+ * #createPartitions(Identifier, List)} does and drops the report, however
much of it the
+ * catalog could have stored, and for a catalog that keeps no partitions
at all that means it
+ * does nothing.
*
* @param identifier path of the table to create partitions
* @param partitions partitions to be created
* @param ignoreIfExists if false, fail when any partition already exists
and apply none of the
* batch; if true, behave like {@link #createPartitions(Identifier,
List)}
+ * @param statistics statistics to report, 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
* @throws TableNotExistException if the table does not exist
+ * @throws UnsupportedOperationException if {@code ignoreIfExists} is
false and the catalog does
+ * not implement strict creation, which is what the default here does
*/
default 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 {
if (!ignoreIfExists) {
throw new UnsupportedOperationException(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
index 13faef105c..6c691e6cee 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -327,9 +327,14 @@ public abstract class DelegateCatalog implements Catalog {
@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);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index f1b5d94cc1..a24fb23549 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -747,15 +747,20 @@ public class RESTCatalog implements Catalog {
@Override
public void createPartitions(Identifier identifier, List<Map<String,
String>> partitions)
throws TableNotExistException {
- createPartitions(identifier, partitions, true);
+ createPartitions(identifier, partitions, true, null, false);
}
@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 {
try {
- api.createPartitions(identifier, partitions, ignoreIfExists);
+ api.createPartitions(
+ identifier, partitions, ignoreIfExists, statistics,
replaceStatistics);
} catch (NoSuchResourceException e) {
throw new TableNotExistException(identifier);
} catch (ForbiddenException e) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
index 4d1009f196..1f4d4575a4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
@@ -23,6 +23,7 @@ import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.utils.FunctionWithException;
import org.apache.paimon.utils.StringUtils;
@@ -31,6 +32,7 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -58,7 +60,7 @@ class CatalogFormatTablePartitionManager implements
FormatTablePartitionManager
CatalogFormatTablePartitionManager(
Identifier identifier, List<String> partitionKeys, CatalogLoader
catalogLoader) {
this.identifier = identifier;
- // Copied: the caller's list must not be able to change what counts as
a leading prefix.
+ // Copied so the caller cannot change what counts as a leading prefix.
this.partitionKeys = Collections.unmodifiableList(new
ArrayList<>(partitionKeys));
this.catalogLoader = catalogLoader;
}
@@ -107,8 +109,7 @@ class CatalogFormatTablePartitionManager implements
FormatTablePartitionManager
do {
PagedList<Partition> page = pageSupplier.page(pageToken);
if (page == null) {
- // A missing page cannot be told apart from an empty listing;
failing loudly
- // beats silently reading fewer partitions.
+ // A missing page cannot be told apart from an empty listing,
so fail loudly.
throw new IllegalStateException(
String.format(
"Catalog returned a null partition page for
format table %s.",
@@ -154,7 +155,14 @@ class CatalogFormatTablePartitionManager implements
FormatTablePartitionManager
}
@Override
- public void createPartitions(List<Map<String, String>> partitions, boolean
ignoreIfExists) {
+ public void createPartitions(
+ List<Map<String, String>> partitions,
+ boolean ignoreIfExists,
+ @Nullable List<PartitionStatistics> statistics,
+ boolean replaceStatistics) {
+ // Validated before the empty check: returning early would swallow a
malformed report.
+ Map<Map<String, String>, PartitionStatistics> statisticsBySpec =
+ validateAndIndexStatistics(statistics, partitions);
if (partitions.isEmpty()) {
return;
}
@@ -163,19 +171,89 @@ class CatalogFormatTablePartitionManager implements
FormatTablePartitionManager
if (!ignoreIfExists) {
// Rejecting the whole batch when any partition exists
is only meaningful
// if the batch stays one request, so a strict create
is never split.
- catalog.createPartitions(identifier, partitions,
false);
+ catalog.createPartitions(
+ identifier, partitions, false, statistics,
replaceStatistics);
return null;
}
- // An idempotent create is safe to split: a rerun
converges from a partially
- // applied batch.
+ // isRetrySafe() bounds the transport retry only: a
caller-level rerun of a
+ // multi-batch ADD still double counts the batches that
already landed.
for (List<Map<String, String>> batch :
batches(partitions)) {
- catalog.createPartitions(identifier, batch, true);
+ // A partition and its statistics travel in the same
request.
+ catalog.createPartitions(
+ identifier,
+ batch,
+ true,
+ statisticsOf(batch, statisticsBySpec),
+ replaceStatistics);
}
return null;
},
"create partitions");
}
+ /**
+ * Indexes the reported statistics by the partition they describe,
rejecting any that describes
+ * a partition this call does not register: a spec typo would otherwise
account for nothing, or
+ * for the wrong partition.
+ *
+ * <p>A spec that appears twice in one request is rejected too, in the
partitions as much as in
+ * the report: batching would apply the repeats once, twice, or once each.
Registration alone is
+ * an idempotent upsert, so a repeat there is harmless.
+ */
+ @Nullable
+ private Map<Map<String, String>, PartitionStatistics>
validateAndIndexStatistics(
+ @Nullable List<PartitionStatistics> statistics, List<Map<String,
String>> partitions) {
+ if (statistics == null) {
+ return null;
+ }
+ // Taken once: a repair of a pre-existing table reaches here with
every partition of the
+ // table, and getFullName() formats a string on every call.
+ String tableName = identifier.getFullName();
+ Set<Map<String, String>> registered = capacityFor(partitions.size());
+ for (Map<String, String> spec : partitions) {
+ checkArgument(
+ registered.add(spec),
+ "Partition %s of table %s is registered twice in one
request that reports "
+ + "statistics; report each partition once.",
+ spec,
+ tableName);
+ }
+ Map<Map<String, String>, PartitionStatistics> bySpec =
+ new HashMap<>(hashCapacity(statistics.size()));
+ for (PartitionStatistics statistic : statistics) {
+ checkArgument(
+ registered.contains(statistic.spec()),
+ "Statistics were reported for partition %s of table %s,
which this request "
+ + "does not register.",
+ statistic.spec(),
+ tableName);
+ checkArgument(
+ bySpec.put(statistic.spec(), statistic) == null,
+ "Statistics were reported twice for partition %s of table
%s; report each "
+ + "partition once.",
+ statistic.spec(),
+ tableName);
+ }
+ return bySpec;
+ }
+
+ @Nullable
+ private static List<PartitionStatistics> statisticsOf(
+ List<Map<String, String>> batch,
+ @Nullable Map<Map<String, String>, PartitionStatistics>
statisticsBySpec) {
+ if (statisticsBySpec == null) {
+ return null;
+ }
+ List<PartitionStatistics> ofBatch = new ArrayList<>(batch.size());
+ for (Map<String, String> spec : batch) {
+ PartitionStatistics statistic = statisticsBySpec.get(spec);
+ if (statistic != null) {
+ ofBatch.add(statistic);
+ }
+ }
+ return ofBatch;
+ }
+
@Override
public void dropPartitions(List<Map<String, String>> partitions) {
if (partitions.isEmpty()) {
@@ -193,6 +271,15 @@ class CatalogFormatTablePartitionManager implements
FormatTablePartitionManager
"drop partitions");
}
+ private static <T> Set<T> capacityFor(int size) {
+ return new HashSet<>(hashCapacity(size));
+ }
+
+ /** Room for {@code size} entries without a rehash, at the default load
factor. */
+ private static int hashCapacity(int size) {
+ return (int) (size / 0.75f) + 1;
+ }
+
private static boolean matchesPrefix(Partition partition, Map<String,
String> prefix) {
for (Map.Entry<String, String> entry : prefix.entrySet()) {
if
(!entry.getValue().equals(partition.spec().get(entry.getKey()))) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
index 278e4c4f36..152baa3ba0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
@@ -166,8 +166,8 @@ public class FormatTableCommit implements BatchTableCommit {
committer.clean(this.fileIO);
}
if (partitionManager != null && !partitionSpecs.isEmpty()) {
- // Concurrent writers may touch the same partition, so
registration is an
- // idempotent ADD rather than a strict create.
+ // Concurrent writers may touch the same partition, so
registration ignores the
+ // ones that already exist rather than failing the commit.
partitionManager.createPartitions(new
ArrayList<>(partitionSpecs), true);
}
for (Map<String, String> partitionSpec : partitionSpecs) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
index 259f88bbfb..0e2c5c9385 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
@@ -22,6 +22,7 @@ import org.apache.paimon.annotation.Experimental;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
import javax.annotation.Nullable;
@@ -61,10 +62,31 @@ public interface FormatTablePartitionManager extends
Serializable {
List<Partition> listPartitionsByNames(List<Map<String, String>>
partitions);
/**
- * Register partitions. With {@code ignoreIfExists=false} the whole batch
is rejected when any
- * partition already exists, so such a request is never split.
+ * Register partitions, reporting no statistics for them. With {@code
ignoreIfExists=false} the
+ * whole batch is rejected when any partition already exists, so such a
request is never split.
*/
- void createPartitions(List<Map<String, String>> partitions, boolean
ignoreIfExists);
+ default void createPartitions(List<Map<String, String>> partitions,
boolean ignoreIfExists) {
+ createPartitions(partitions, ignoreIfExists, null, false);
+ }
+
+ /**
+ * Register partitions and report statistics for them in the same call, so
a partition is never
+ * registered by a request whose statistics failed on their own.
+ *
+ * <p>Statistics are matched to {@code partitions} by {@link
PartitionStatistics#spec()} and may
+ * cover only some of them; {@code replaceStatistics} says whether they
replace what the catalog
+ * holds or add to it, and is ignored when {@code statistics} is null.
Reporting never
+ * unregisters a partition.
+ *
+ * <p>This is the method an implementation provides, so that none can
report nothing by
+ * accident: a decorator that forwards only the two-argument form would
otherwise drop every
+ * report and leave the caller no way to notice.
+ */
+ void createPartitions(
+ List<Map<String, String>> partitions,
+ boolean ignoreIfExists,
+ @Nullable List<PartitionStatistics> statistics,
+ boolean replaceStatistics);
/** Unregister partitions. Metadata only; missing partitions are ignored.
*/
void dropPartitions(List<Map<String, String>> partitions);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
index 0961050f11..37b991fc68 100644
---
a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
@@ -25,6 +25,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.Table;
@@ -363,11 +364,34 @@ class CachingCatalogTest extends CatalogTestBase {
when(wrapped.listPartitions(identifier)).thenReturn(emptyList(),
singletonList(created));
assertThat(catalog.listPartitions(identifier)).isEmpty();
- catalog.createPartitions(identifier, singletonList(spec), false);
+ catalog.createPartitions(identifier, singletonList(spec), false, null,
false);
assertThat(catalog.listPartitions(identifier)).containsExactly(created);
}
+ @Test
+ public void
testCreatePartitionsWithStatisticsForwardsAndInvalidatesPartitionCache()
+ throws Exception {
+ Catalog wrapped = Mockito.mock(Catalog.class);
+ TestableCachingCatalog catalog =
+ new TestableCachingCatalog(wrapped, EXPIRATION_TTL, ticker);
+ Identifier identifier = new Identifier("db", "tbl");
+ Map<String, String> spec = singletonMap("dt", "20260717");
+ Partition created = new Partition(spec, 3, 300, 1, 1000, -1, false);
+ List<PartitionStatistics> statistics =
+ singletonList(new PartitionStatistics(spec, 3, 300, 1, 1000,
-1));
+ when(wrapped.listPartitions(identifier)).thenReturn(emptyList(),
singletonList(created));
+
+ assertThat(catalog.listPartitions(identifier)).isEmpty();
+ catalog.createPartitions(identifier, singletonList(spec), true,
statistics, false);
+
+ // Dropping the forward would leave the statistics unreported and
nothing else would say so.
+ Mockito.verify(wrapped)
+ .createPartitions(identifier, singletonList(spec), true,
statistics, false);
+ // A report changes what a partition holds, so the cached listing is
stale after it.
+
assertThat(catalog.listPartitions(identifier)).containsExactly(created);
+ }
+
@Test
public void testDeadlock() throws Exception {
Catalog underlyCatalog = this.catalog;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java
new file mode 100644
index 0000000000..ec230c8504
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.catalog;
+
+import org.apache.paimon.partition.PartitionStatistics;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Tests for {@link DelegateCatalog}. A missing forward does not fail: the
interface default drops
+ * down to the call that predates statistics, so only the report disappears.
+ */
+class DelegateCatalogTest {
+
+ private static final Identifier IDENTIFIER = Identifier.create("db", "t");
+
+ @Test
+ void testCreatePartitionsCarriesStatisticsAndModeToTheWrappedCatalog()
throws Exception {
+ Catalog wrapped = mock(Catalog.class);
+ Catalog delegating = new TestDelegateCatalog(wrapped);
+ List<Map<String, String>> specs =
+ Arrays.asList(
+ Collections.singletonMap("dt", "20260728"),
+ Collections.singletonMap("dt", "20260729"));
+ List<PartitionStatistics> statistics =
+ Collections.singletonList(
+ new PartitionStatistics(specs.get(0), 3L, 300L, 1L,
1000L, -1));
+
+ delegating.createPartitions(IDENTIFIER, specs, true, statistics,
false);
+
+ verify(wrapped).createPartitions(IDENTIFIER, specs, true, statistics,
false);
+ // Falling through to the two-argument call is how the statistics
would go missing.
+ verify(wrapped, never()).createPartitions(any(), anyList());
+ }
+
+ @Test
+ void testCreatePartitionsCarriesTheAbsenceOfStatisticsThrough() throws
Exception {
+ Catalog wrapped = mock(Catalog.class);
+ Catalog delegating = new TestDelegateCatalog(wrapped);
+ List<Map<String, String>> specs =
+ Collections.singletonList(Collections.singletonMap("dt",
"20260728"));
+
+ delegating.createPartitions(IDENTIFIER, specs, false, null, false);
+
+ verify(wrapped).createPartitions(IDENTIFIER, specs, false, null,
false);
+ }
+
+ /** {@link DelegateCatalog} forwards every operation; these tests never
rebuild one. */
+ private static class TestDelegateCatalog extends DelegateCatalog {
+
+ TestDelegateCatalog(Catalog wrapped) {
+ super(wrapped);
+ }
+
+ @Override
+ public CatalogLoader catalogLoader() {
+ return wrapped.catalogLoader();
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
index a68b7933e5..093c8f7bf5 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
@@ -31,6 +31,7 @@ import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -303,6 +304,165 @@ class MockRESTCatalogTest extends RESTCatalogTest {
.containsExactly(partition);
}
+ @Test
+ void testReportedPartitionStatisticsAreStoredAndReadBack() throws
Exception {
+ Identifier identifier =
createFormatTableWithCatalogManagedPartitions();
+ Map<String, String> spec = Collections.singletonMap("dt", "20260717");
+ List<Map<String, String>> specs = Collections.singletonList(spec);
+ FormatTablePartitionManager partitionManager =
+ ((FormatTable)
restCatalog.getTable(identifier)).partitionManager();
+ assertThat(partitionManager).isNotNull();
+
+ // A registration on its own measures nothing, so everything starts
out unknown.
+ restCatalog.createPartitions(identifier, specs);
+ Partition registered = onlyPartition(identifier);
+
assertThat(PartitionStatistics.isKnown(registered.recordCount())).isFalse();
+
assertThat(PartitionStatistics.isKnown(registered.fileCount())).isFalse();
+
+ // ADD onto a partition nobody measured yet: the report becomes what
it holds.
+ restCatalog
+ .api()
+ .createPartitions(
+ identifier,
+ specs,
+ true,
+ Collections.singletonList(
+ new PartitionStatistics(spec, 3L, 300L, 1L,
1000L, -1)),
+ false);
+ assertStatistics(identifier, 3L, 300L, 1L, 1000L);
+
+ // ADD again, through the partition manager a writer commits with: the
counts accumulate
+ // and an older file does not move the newest one backwards.
+ partitionManager.createPartitions(
+ specs,
+ true,
+ Collections.singletonList(new PartitionStatistics(spec, 4L,
400L, 2L, 500L, -1)),
+ false);
+ assertStatistics(identifier, 7L, 700L, 3L, 1000L);
+
+ // A field reported as unknown leaves the stored one alone rather than
zeroing it.
+ restCatalog.createPartitions(
+ identifier,
+ specs,
+ true,
+ Collections.singletonList(
+ new PartitionStatistics(
+ spec,
+ PartitionStatistics.UNKNOWN,
+ 100L,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN,
+ -1)),
+ false);
+ assertStatistics(identifier, 7L, 800L, 3L, 1000L);
+
+ // SET is the whole partition now: every reported field is replaced,
including a creation
+ // time that moves backwards because the newer files are gone.
+ restCatalog
+ .api()
+ .createPartitions(
+ identifier,
+ specs,
+ true,
+ Collections.singletonList(
+ new PartitionStatistics(spec, 5L, 500L, 1L,
700L, -1)),
+ true);
+ assertStatistics(identifier, 5L, 500L, 1L, 700L);
+
+ // Unknown is skipped under SET too: it reports nothing about that
field, not a zero.
+ restCatalog.createPartitions(
+ identifier,
+ specs,
+ true,
+ Collections.singletonList(
+ new PartitionStatistics(
+ spec,
+ PartitionStatistics.UNKNOWN,
+ 900L,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN,
+ -1)),
+ true);
+ assertStatistics(identifier, 5L, 900L, 1L, 700L);
+
+ // Reporting never registers or unregisters anything.
+ assertThat(restCatalog.listPartitions(identifier)).hasSize(1);
+ }
+
+ @Test
+ void testStatisticsOfAnUnstoredPartitionAreDropped() throws Exception {
+ Identifier identifier =
createFormatTableWithCatalogManagedPartitions();
+ Map<String, String> spec = Collections.singletonMap("dt", "20260717");
+
+ // The statistics describe a partition this request does not register,
so the server drops
+ // them and keeps the registration.
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(
+ new PartitionStatistics(
+ Collections.singletonMap("dt", "20260718"),
+ 9L,
+ 900L,
+ 3L,
+ 1000L,
+ -1)),
+ false);
+
+ assertThat(restCatalog.listPartitions(identifier))
+ .extracting(Partition::spec)
+ .containsExactly(spec);
+
assertThat(PartitionStatistics.isKnown(onlyPartition(identifier).recordCount())).isFalse();
+ }
+
+ @Test
+ void testAReportThatOnlyPartlyMatchesIsNotAppliedAtAll() throws Exception {
+ Identifier identifier =
createFormatTableWithCatalogManagedPartitions();
+ Map<String, String> stored = Collections.singletonMap("dt",
"20260717");
+ Map<String, String> absent = Collections.singletonMap("dt",
"20260718");
+ restCatalog.createPartitions(identifier,
Collections.singletonList(stored));
+
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(stored),
+ true,
+ Arrays.asList(
+ new PartitionStatistics(stored, 3L, 300L, 1L, 1000L,
-1),
+ new PartitionStatistics(absent, 9L, 900L, 3L, 2000L,
-1)),
+ false);
+
+ // Applying the half that matched would count it twice on the next
report.
+ Partition partition = onlyPartition(identifier);
+
assertThat(PartitionStatistics.isKnown(partition.recordCount())).isFalse();
+
assertThat(PartitionStatistics.isKnown(partition.fileSizeInBytes())).isFalse();
+
assertThat(PartitionStatistics.isKnown(partition.fileCount())).isFalse();
+
assertThat(PartitionStatistics.isKnown(partition.lastFileCreationTime())).isFalse();
+ }
+
+ private Partition onlyPartition(Identifier identifier) throws Exception {
+ List<Partition> partitions = restCatalog.listPartitions(identifier);
+ assertThat(partitions).hasSize(1);
+ return partitions.get(0);
+ }
+
+ private void assertStatistics(
+ Identifier identifier,
+ long recordCount,
+ long fileSizeInBytes,
+ long fileCount,
+ long lastFileCreationTime)
+ throws Exception {
+ Partition partition = onlyPartition(identifier);
+ assertThat(
+ Arrays.asList(
+ partition.recordCount(),
+ partition.fileSizeInBytes(),
+ partition.fileCount(),
+ partition.lastFileCreationTime()))
+ .containsExactly(recordCount, fileSizeInBytes, fileCount,
lastFileCreationTime);
+ }
+
@Test
void testFilteredListingPreservesNextTokenAcrossSparsePage() throws
Exception {
Identifier identifier =
createFormatTableWithCatalogManagedPartitions();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTApiJsonTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTApiJsonTest.java
index 4a8b4ed964..44449fcabc 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTApiJsonTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTApiJsonTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.rest;
import org.apache.paimon.function.FunctionChange;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.rest.requests.AlterDatabaseRequest;
import org.apache.paimon.rest.requests.AlterFunctionRequest;
import org.apache.paimon.rest.requests.AlterTableRequest;
@@ -64,6 +65,7 @@ import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Test for {@link RESTApi} json. */
@@ -326,6 +328,53 @@ public class RESTApiJsonTest {
CreatePartitionsRequest parsedExplicitRequest =
RESTApi.fromJson(explicitRequestJson,
CreatePartitionsRequest.class);
assertFalse(parsedExplicitRequest.ignoreIfExists());
+
+ // A client that reports nothing sends neither field, so an older
server sees exactly the
+ // request it saw before.
+ assertFalse(explicitRequestJson.contains("partitionStatistics"));
+ assertFalse(explicitRequestJson.contains("replaceStatistics"));
+ assertNull(defaultRequest.getPartitionStatistics());
+ assertNull(defaultRequest.replaceStatistics());
+ }
+
+ @Test
+ public void createPartitionsRequestCarriesStatisticsTest() throws
Exception {
+ Map<String, String> spec = Collections.singletonMap("dt", "20260728");
+ PartitionStatistics statistics =
+ new PartitionStatistics(spec, 7L, 4096L, 2L, 1753660800000L,
-1);
+ CreatePartitionsRequest request =
+ new CreatePartitionsRequest(
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(statistics),
+ true);
+
+ String json = RESTApi.toJson(request);
+ // Asserted literally: renaming both ends together would keep every
parse below passing.
+ assertTrue(json.contains("\"partitionStatistics\""));
+ assertTrue(json.contains("\"replaceStatistics\":true"));
+ // How the client treats its own request; the server is not told and
would reject the
+ // field. It is a getter on a serialized type, so it only stays off
the wire on purpose.
+ assertFalse(json.contains("retrySafe"));
+
+ CreatePartitionsRequest parsed = RESTApi.fromJson(json,
CreatePartitionsRequest.class);
+ assertEquals(Boolean.TRUE, parsed.replaceStatistics());
+ assertEquals(Collections.singletonList(statistics),
parsed.getPartitionStatistics());
+
+ // An unknown field stays unknown across the wire rather than turning
into a zero.
+ PartitionStatistics unknown = PartitionStatistics.unknown(spec);
+ CreatePartitionsRequest unknownRequest =
+ new CreatePartitionsRequest(
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(unknown),
+ false);
+ PartitionStatistics parsedUnknown =
+ RESTApi.fromJson(RESTApi.toJson(unknownRequest),
CreatePartitionsRequest.class)
+ .getPartitionStatistics()
+ .get(0);
+ assertFalse(PartitionStatistics.isKnown(parsedUnknown.recordCount()));
+ assertFalse(PartitionStatistics.isKnown(parsedUnknown.fileCount()));
}
@Test
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
index c571b19c38..733b4b1c9f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -1942,18 +1942,114 @@ public class RESTCatalogServer {
List<Map<String, String>> existed = new ArrayList<>();
for (Map<String, String> spec : request.getPartitionSpecs()) {
if (existingSpecs.add(spec)) {
- storedPartitions.add(new Partition(spec, 0, 0, 0, 0,
-1, false));
+ // A registration measures nothing, so a new partition
starts unknown.
+ storedPartitions.add(
+ new Partition(
+ spec,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN,
+
PartitionStatistics.UNKNOWN_TOTAL_BUCKETS,
+ false));
created.add(spec);
} else {
existed.add(spec);
}
}
+ applyPartitionStatistics(
+ storedPartitions,
+ request.getPartitionStatistics(),
+ request.replaceStatistics());
return mockResponse(new CreatePartitionsResponse(created,
existed), 200);
default:
return new MockResponse().setResponseCode(404);
}
}
+ /**
+ * Folds reported statistics into the stored partitions, the way a catalog
server does:
+ * replacing overwrites, adding accumulates, a field reported as unknown
leaves the stored one
+ * alone, and no report adds or removes a partition row.
+ *
+ * <p>All or nothing: if any reported spec names a partition this table
does not hold, none of
+ * the report is applied, since a reporter sending it again would count
the applied part twice.
+ */
+ private static void applyPartitionStatistics(
+ List<Partition> storedPartitions,
+ @Nullable List<PartitionStatistics> statistics,
+ @Nullable Boolean replaceStatistics) {
+ if (statistics == null) {
+ return;
+ }
+ boolean accumulate = !Boolean.TRUE.equals(replaceStatistics);
+ Map<Map<String, String>, PartitionStatistics> reported = new
HashMap<>();
+ for (PartitionStatistics statistic : statistics) {
+ reported.put(statistic.spec(), statistic);
+ }
+ Set<Map<String, String>> storedSpecs =
+
storedPartitions.stream().map(Partition::spec).collect(Collectors.toSet());
+ if (!storedSpecs.containsAll(reported.keySet())) {
+ // Applying the half that matched would count it twice on the next
report.
+ return;
+ }
+ for (int i = 0; i < storedPartitions.size(); i++) {
+ Partition stored = storedPartitions.get(i);
+ PartitionStatistics update = reported.get(stored.spec());
+ if (update == null) {
+ continue;
+ }
+ storedPartitions.set(
+ i,
+ new Partition(
+ stored.spec(),
+ combine(stored.recordCount(),
update.recordCount(), accumulate),
+ combine(stored.fileSizeInBytes(),
update.fileSizeInBytes(), accumulate),
+ combine(stored.fileCount(), update.fileCount(),
accumulate),
+ combineLastFileCreationTime(
+ stored.lastFileCreationTime(),
+ update.lastFileCreationTime(),
+ accumulate),
+ stored.totalBuckets(),
+ stored.done()));
+ }
+ }
+
+ /**
+ * Folds a snapshot commit's report onto a stored value. That report is a
delta, so a negative
+ * is a decrement rather than an unknown, but a value nobody has measured
is replaced rather
+ * than added to: a partition registered and not yet measured holds
UNKNOWN, and UNKNOWN plus a
+ * count is a count short by one.
+ */
+ private static long accumulateDelta(long stored, long reported) {
+ return PartitionStatistics.isKnown(stored) ? stored + reported :
reported;
+ }
+
+ private static long combine(long stored, long reported, boolean
accumulate) {
+ if (!PartitionStatistics.isKnown(reported)) {
+ return stored;
+ }
+ if (!accumulate || !PartitionStatistics.isKnown(stored)) {
+ return reported;
+ }
+ return stored + reported;
+ }
+
+ /**
+ * Folds a reported creation time in: adding takes the later of the two,
setting takes what the
+ * report says even when that moves the time backwards.
+ */
+ private static long combineLastFileCreationTime(
+ long stored, long reported, boolean accumulate) {
+ if (!PartitionStatistics.isKnown(reported)) {
+ return stored;
+ }
+ if (!accumulate) {
+ return reported;
+ }
+ return Math.max(stored, reported);
+ }
+
private MockResponse dropPartitionsHandle(String data, Identifier
tableIdentifier)
throws Exception {
DropPartitionsRequest request = RESTApi.fromJson(data,
DropPartitionsRequest.class);
@@ -2868,12 +2964,16 @@ public class RESTCatalogServer {
}
return new Partition(
oldPartition.spec(),
-
oldPartition.recordCount()
- +
stats.recordCount(),
-
oldPartition.fileSizeInBytes()
- +
stats.fileSizeInBytes(),
-
oldPartition.fileCount()
- +
stats.fileCount(),
+
accumulateDelta(
+
oldPartition.recordCount(),
+
stats.recordCount()),
+
accumulateDelta(
+
oldPartition
+
.fileSizeInBytes(),
+
stats.fileSizeInBytes()),
+
accumulateDelta(
+
oldPartition.fileCount(),
+
stats.fileCount()),
Math.max(
oldPartition
.lastFileCreationTime(),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index 8fc5e38a25..87116e3e45 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -1567,7 +1567,7 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
List<Map<String, String>> partitionSpecs =
Arrays.asList(singletonMap("dt", "20260714"),
singletonMap("dt", "20260715"));
CreatePartitionsResponse response =
- restCatalog.api().createPartitions(identifier, partitionSpecs);
+ restCatalog.api().createPartitions(identifier, partitionSpecs,
true, null, false);
assertThat(response.getCreated()).containsExactlyInAnyOrderElementsOf(partitionSpecs);
assertThat(response.getExisted()).isEmpty();
@@ -1581,7 +1581,8 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
() ->
restCatalog
.api()
- .createPartitions(identifier,
conflictingSpecs, false))
+ .createPartitions(
+ identifier, conflictingSpecs,
false, null, false))
.isInstanceOf(AlreadyExistsException.class)
.hasMessageContaining("dt=20260714");
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java
index dfcc422be3..0c558973b7 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java
@@ -24,6 +24,7 @@ import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.types.DataType;
@@ -446,6 +447,188 @@ class CatalogFormatTablePartitionManagerTest {
verifyNoInteractions(listCatalog);
}
+ // ------------------------------------------------------------------------
+ // reported statistics
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testStatisticsRideInTheRequestOfTheirOwnPartitions() throws Exception
{
+ Catalog catalog = mock(Catalog.class);
+ List<Map<String, String>> specs = specs(2500);
+ // Astride both split points, where a partition and its statistics
could come apart.
+ List<PartitionStatistics> statistics =
+ Arrays.asList(
+ statistics(specs.get(0), 1L),
+ statistics(specs.get(999), 2L),
+ statistics(specs.get(1000), 3L),
+ statistics(specs.get(2499), 4L));
+
+ partitionManager(catalog).createPartitions(specs, true, statistics,
false);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<Map<String, String>>> specCaptor =
ArgumentCaptor.forClass(List.class);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<PartitionStatistics>> statisticsCaptor =
+ ArgumentCaptor.forClass(List.class);
+ verify(catalog, times(3))
+ .createPartitions(
+ eq(IDENTIFIER),
+ specCaptor.capture(),
+ eq(true),
+ statisticsCaptor.capture(),
+ eq(false));
+ List<List<Map<String, String>>> requestedSpecs =
specCaptor.getAllValues();
+ List<List<PartitionStatistics>> requestedStatistics =
statisticsCaptor.getAllValues();
+
assertThat(requestedSpecs).extracting(List::size).containsExactly(1000, 1000,
500);
+ assertThat(flatten(requestedSpecs)).isEqualTo(specs);
+ for (int request = 0; request < requestedSpecs.size(); request++) {
+ List<Map<String, String>> registered = requestedSpecs.get(request);
+ for (PartitionStatistics statistic :
requestedStatistics.get(request)) {
+ assertThat(registered).contains(statistic.spec());
+ }
+ }
+ // Split apart, never dropped or duplicated.
+
assertThat(requestedStatistics).extracting(List::size).containsExactly(2, 1, 1);
+
assertThat(requestedStatistics.stream().flatMap(List::stream).collect(Collectors.toList()))
+ .containsExactlyElementsOf(statistics);
+ }
+
+ @Test
+ void testABatchThatReportsNothingSendsAnEmptyListNotNull() throws
Exception {
+ Catalog catalog = mock(Catalog.class);
+ List<Map<String, String>> specs = specs(2500);
+ // All in the first request, so the two that follow are the ones that
report nothing.
+ List<PartitionStatistics> statistics =
+ Arrays.asList(statistics(specs.get(0), 1L),
statistics(specs.get(999), 2L));
+
+ partitionManager(catalog).createPartitions(specs, true, statistics,
false);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<PartitionStatistics>> statisticsCaptor =
+ ArgumentCaptor.forClass(List.class);
+ verify(catalog, times(3))
+ .createPartitions(
+ eq(IDENTIFIER), anyList(), eq(true),
statisticsCaptor.capture(), eq(false));
+ List<List<PartitionStatistics>> requestedStatistics =
statisticsCaptor.getAllValues();
+
assertThat(requestedStatistics.get(0)).containsExactlyElementsOf(statistics);
+ // Null would mean "this client does not report", which is the call
that predates
+ // statistics; empty means "this request measures nothing", which is
what happened.
+ assertThat(requestedStatistics.get(1)).isNotNull().isEmpty();
+ assertThat(requestedStatistics.get(2)).isNotNull().isEmpty();
+ }
+
+ @Test
+ void testStrictCreateWithStatisticsStaysOneRequest() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ List<Map<String, String>> specs = specs(2500);
+ List<PartitionStatistics> statistics =
+ Arrays.asList(statistics(specs.get(0), 1L),
statistics(specs.get(2499), 2L));
+
+ partitionManager(catalog).createPartitions(specs, false, statistics,
true);
+
+ verify(catalog).createPartitions(IDENTIFIER, specs, false, statistics,
true);
+ }
+
+ @Test
+ void testStatisticsForAnUnregisteredPartitionAreRejected() {
+ Catalog catalog = mock(Catalog.class);
+ FormatTablePartitionManager partitionManager =
partitionManager(catalog);
+ List<Map<String, String>> specs =
Collections.singletonList(spec("2025", "01"));
+ // A spec typo would otherwise account for nothing at all, silently.
+ List<PartitionStatistics> statistics =
+ Collections.singletonList(statistics(spec("2025", "02"), 7L));
+
+ assertThatThrownBy(() -> partitionManager.createPartitions(specs,
true, statistics, false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("does not register")
+ .hasMessageContaining("month=02")
+
.hasMessageContaining("catalog_partition_db.catalog_partition_table");
+
+ verifyNoInteractions(catalog);
+ }
+
+ @Test
+ void testStatisticsReportedWithNoPartitionsAreRejected() {
+ Catalog catalog = mock(Catalog.class);
+ FormatTablePartitionManager partitionManager =
partitionManager(catalog);
+ // Nothing is registered here, so every reported spec is one this
request does not
+ // register. Returning quietly because the list is empty would swallow
the whole report.
+ List<PartitionStatistics> statistics =
+ Collections.singletonList(statistics(spec("2025", "01"), 7L));
+
+ assertThatThrownBy(
+ () ->
+ partitionManager.createPartitions(
+ Collections.emptyList(), true,
statistics, false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("does not register")
+
.hasMessageContaining("catalog_partition_db.catalog_partition_table");
+
+ verifyNoInteractions(catalog);
+ }
+
+ @Test
+ void testAWellFormedEmptyReportWithNoPartitionsTouchesNoCatalog() {
+ // Validation comes first, but a report that describes nothing and
registers nothing is
+ // still nothing to send.
+ Catalog catalog = mock(Catalog.class);
+
+ partitionManager(catalog)
+ .createPartitions(Collections.emptyList(), true,
Collections.emptyList(), false);
+
+ verifyNoInteractions(catalog);
+ }
+
+ @Test
+ void testRepeatedPartitionWithStatisticsIsRejected() {
+ Catalog catalog = mock(Catalog.class);
+ FormatTablePartitionManager partitionManager =
partitionManager(catalog);
+ Map<String, String> repeated = spec("2025", "01");
+ List<Map<String, String>> specs =
+ Arrays.asList(repeated, spec("2025", "02"), spec("2025",
"01"));
+ List<PartitionStatistics> statistics =
Collections.singletonList(statistics(repeated, 7L));
+
+ assertThatThrownBy(() -> partitionManager.createPartitions(specs,
true, statistics, false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("registered twice")
+ .hasMessageContaining("month=01")
+
.hasMessageContaining("catalog_partition_db.catalog_partition_table");
+
+ verifyNoInteractions(catalog);
+ }
+
+ @Test
+ void testRepeatedStatisticsForOnePartitionAreRejected() {
+ Catalog catalog = mock(Catalog.class);
+ FormatTablePartitionManager partitionManager =
partitionManager(catalog);
+ Map<String, String> spec = spec("2025", "01");
+ List<PartitionStatistics> statistics =
+ Arrays.asList(statistics(spec, 7L), statistics(spec, 9L));
+
+ assertThatThrownBy(
+ () ->
+ partitionManager.createPartitions(
+ Collections.singletonList(spec), true,
statistics, false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("reported twice")
+ .hasMessageContaining("month=01")
+
.hasMessageContaining("catalog_partition_db.catalog_partition_table");
+
+ verifyNoInteractions(catalog);
+ }
+
+ @Test
+ void testRepeatedPartitionWithoutStatisticsIsAccepted() throws Exception {
+ // A bare registration is an idempotent upsert: repeating one
registers the same partition
+ // again and changes nothing, so it is not worth failing a commit over.
+ Catalog catalog = mock(Catalog.class);
+ List<Map<String, String>> specs = Arrays.asList(spec("2025", "01"),
spec("2025", "01"));
+
+ partitionManager(catalog).createPartitions(specs, true);
+
+ assertThat(capturedCreates(catalog, true, 1).get(0)).isEqualTo(specs);
+ }
+
// ------------------------------------------------------------------------
// catalog lifecycle
// ------------------------------------------------------------------------
@@ -523,7 +706,9 @@ class CatalogFormatTablePartitionManagerTest {
void testRuntimeExceptionIsRethrownUnchanged() throws Exception {
Catalog catalog = mock(Catalog.class);
RuntimeException failure = new IllegalStateException("catalog
unavailable");
- doThrow(failure).when(catalog).createPartitions(any(), anyList(),
anyBoolean());
+ doThrow(failure)
+ .when(catalog)
+ .createPartitions(any(), anyList(), anyBoolean(), any(),
anyBoolean());
FormatTablePartitionManager partitionManager =
partitionManager(catalog);
Throwable thrown = catchThrowable(() ->
partitionManager.createPartitions(specs(1), true));
@@ -559,12 +744,14 @@ class CatalogFormatTablePartitionManagerTest {
return FormatTablePartitionManager.create(IDENTIFIER, PARTITION_KEYS,
() -> catalog);
}
+ /** The specs of every request a create that reports nothing sent. */
private static List<List<Map<String, String>>> capturedCreates(
Catalog catalog, boolean ignoreIfExists, int expectedRequests)
throws Exception {
@SuppressWarnings("unchecked")
ArgumentCaptor<List<Map<String, String>>> captor =
ArgumentCaptor.forClass(List.class);
verify(catalog, times(expectedRequests))
- .createPartitions(eq(IDENTIFIER), captor.capture(),
eq(ignoreIfExists));
+ .createPartitions(
+ eq(IDENTIFIER), captor.capture(), eq(ignoreIfExists),
isNull(), eq(false));
return captor.getAllValues();
}
@@ -627,6 +814,17 @@ class CatalogFormatTablePartitionManagerTest {
return spec;
}
+ /** Statistics of one partition, told apart by their record count. */
+ private static PartitionStatistics statistics(Map<String, String> spec,
long recordCount) {
+ return new PartitionStatistics(
+ spec,
+ recordCount,
+ recordCount * 1024,
+ 1L,
+ 1753660800000L,
+ PartitionStatistics.UNKNOWN_TOTAL_BUCKETS);
+ }
+
private static List<Map<String, String>> specs(int count) {
List<Map<String, String>> specs = new ArrayList<>(count);
for (int i = 0; i < count; i++) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
index 176dbf8291..56b7a18493 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
@@ -30,6 +30,7 @@ import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.table.FormatTable;
@@ -400,7 +401,10 @@ class CatalogManagedPartitionScanTest {
@Override
public void createPartitions(
- List<Map<String, String>> partitions, boolean
ignoreIfExists) {
+ List<Map<String, String>> partitions,
+ boolean ignoreIfExists,
+ @Nullable List<PartitionStatistics> statistics,
+ boolean replaceStatistics) {
throw new UnsupportedOperationException();
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
index bc6e24d219..033285c46e 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
@@ -25,6 +25,7 @@ import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.format.FormatTablePartitionManager;
@@ -444,7 +445,11 @@ class FormatTablePartitionRepairTest {
}
@Override
- public void createPartitions(List<Map<String, String>> partitions,
boolean ignoreIfExists) {
+ public void createPartitions(
+ List<Map<String, String>> partitions,
+ boolean ignoreIfExists,
+ @Nullable List<PartitionStatistics> statistics,
+ boolean replaceStatistics) {
createdPartitions.add(new ArrayList<>(partitions));
createIgnoreFlags.add(ignoreIfExists);
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/execution/FormatTablePartitionDdlPlanningTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/execution/FormatTablePartitionDdlPlanningTest.scala
index 7968c8d49c..8c24e43247 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/execution/FormatTablePartitionDdlPlanningTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/execution/FormatTablePartitionDdlPlanningTest.scala
@@ -22,7 +22,7 @@ import org.apache.paimon.catalog.{CatalogContext, Identifier}
import org.apache.paimon.fs.{FileIO, Path}
import org.apache.paimon.fs.local.LocalFileIO
import org.apache.paimon.options.Options
-import org.apache.paimon.partition.Partition
+import org.apache.paimon.partition.{Partition, PartitionStatistics}
import org.apache.paimon.predicate.Predicate
import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase
import org.apache.paimon.spark.catalyst.plans.logical.PaimonDropPartitions
@@ -368,7 +368,9 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = dropCalls += 1
@@ -419,7 +421,9 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {
dropped = partitions.asScala.map(_.asScala.toMap).toSeq
@@ -486,7 +490,9 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
def newGateway(): FormatTablePartitionManager = new
FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {
val specs = partitions.asScala.map(_.asScala.toMap).toSeq
compensationCreates :+= ((specs, ignoreIfExists))
registered ++= specs
@@ -551,7 +557,9 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
def newGateway(): FormatTablePartitionManager = new
FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {
val specs = partitions.asScala.map(_.asScala.toMap).toSeq
compensationCreates :+= ((specs, ignoreIfExists))
registered ++= specs
@@ -615,7 +623,9 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {
dropped = partitions.asScala.map(_.asScala.toMap).toSeq
@@ -657,7 +667,9 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {
dropped = partitions.asScala.map(_.asScala.toMap).toSeq
@@ -713,7 +725,9 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = dropCalls += 1
@@ -869,8 +883,7 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
private def registerPartitions(tableName: String, specs: Map[String,
String]*): Unit =
paimonCatalog.createPartitions(
Identifier.create(dbName0, tableName),
- specs.map(_.asJava).asJava,
- true)
+ specs.map(_.asJava).asJava)
private def registeredPartitionSpecs(tableName: String): Set[Map[String,
String]] =
paimonCatalog
@@ -908,7 +921,9 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {
createCalls += 1
created = partitions.asScala.toSeq
this.ignoreIfExists = ignoreIfExists
@@ -941,7 +956,9 @@ class FormatTablePartitionDdlPlanningTest extends
PaimonSparkTestWithRestCatalog
override def createPartitions(
partitionsToCreate: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = synchronized {
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = synchronized {
val batch = partitionsToCreate.asScala.map(_.asScala.toMap).toSeq
batches :+= batch
val duplicates = batch.filter(partitions.contains)
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
index 7c36fd09be..8febee0e0a 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
@@ -22,7 +22,7 @@ import org.apache.paimon.catalog.{CatalogContext, Identifier}
import org.apache.paimon.fs.{FileIO, Path}
import org.apache.paimon.fs.local.LocalFileIO
import org.apache.paimon.options.Options
-import org.apache.paimon.partition.Partition
+import org.apache.paimon.partition.{Partition, PartitionStatistics}
import org.apache.paimon.predicate.Predicate
import org.apache.paimon.table.FormatTable
import org.apache.paimon.table.format.FormatTablePartitionManager
@@ -51,7 +51,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {
forwardedPartitions = partitions.asScala.toSeq
forwardedIgnoreIfExists = ignoreIfExists
}
@@ -108,7 +110,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = dropCalls += 1
@@ -141,7 +145,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = createCalls += 1
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = createCalls += 1
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {}
@@ -173,7 +179,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = createCalls += 1
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = createCalls += 1
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {}
@@ -242,7 +250,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {
// Ordering contract: the directory must still exist when the catalog
unregisters.
@@ -294,7 +304,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {
dropped = partitions.asScala.map(_.asScala.toMap).toSeq
@@ -410,7 +422,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {
dropped = partitions.asScala.map(_.asScala.toMap).toSeq
@@ -453,7 +467,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = dropCalls += 1
@@ -501,7 +517,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
val gateway = new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit =
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit =
throw new AssertionError("An ambiguous DROP must not recreate catalog
partitions")
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {
@@ -611,7 +629,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
new FormatTablePartitionManager {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]):
Unit = {}
@@ -638,7 +658,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = synchronized {
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = synchronized {
val requested = partitions.asScala.map(_.asScala.toMap).toSeq
requests :+= ((requested, ignoreIfExists))
if (!ignoreIfExists) {
@@ -676,7 +698,9 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {}
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {}
override def dropPartitions(partitions: JList[JMap[String, String]]): Unit
= dropCalls += 1
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
index 24e42b9c86..586099b73b 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
@@ -20,7 +20,7 @@ package org.apache.paimon.spark.sql
import org.apache.paimon.catalog.Identifier
import org.apache.paimon.fs.Path
-import org.apache.paimon.partition.Partition
+import org.apache.paimon.partition.{Partition, PartitionStatistics}
import org.apache.paimon.predicate.Predicate
import org.apache.paimon.spark.{PaimonSparkTestWithRestCatalogBase,
SparkCatalog}
import org.apache.paimon.spark.execution.PaimonRepairFormatTablePartitionsExec
@@ -503,8 +503,7 @@ class CatalogManagedPartitionMsckRepairTest extends
PaimonSparkTestWithRestCatal
private def registerPartitions(tableName: String, partitions: String*): Unit
=
paimonCatalog.createPartitions(
tableIdentifier(tableName),
- partitions.map(value => Map("dt" -> value).asJava).asJava,
- true)
+ partitions.map(value => Map("dt" -> value).asJava).asJava)
private def registeredPartitions(tableName: String): Set[String] =
paimonCatalog
@@ -557,7 +556,9 @@ private[sql] class StatefulFaultCatalog(initial:
Set[Map[String, String]] = Set.
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {
createCalls += 1
state ++= partitions.asScala.map(_.asScala.toMap)
if (failCreateAfterApply) {
@@ -644,7 +645,9 @@ private[sql] class
FaultInjectingFormatTablePartitionManager(delegate: FormatTab
override def createPartitions(
partitions: JList[JMap[String, String]],
- ignoreIfExists: Boolean): Unit = {
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean): Unit = {
delegate.createPartitions(partitions, ignoreIfExists)
MsckFaultInjection.createCalls += 1
}
