From 5ea393cadda0059947fb3d5d9fb7e109c6407fd3 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 19 Aug 2026 17:37:29 +0800 Subject: [PATCH] [core][rest] Report partition statistics when registering partitions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A catalog-managed format table registers the partitions a commit wrote. It has no snapshot, so the channel data tables use to report statistics — commitSnapshot — does not exist for it, and there is nowhere else for the numbers to go. This adds them to the registration itself. CreatePartitionsRequest gains an optional partitionStatistics list and a replaceStatistics flag. Registration and statistics then land in one request and one server-side transaction, which is the point: "the statistics failed but the partition registered" is not a state anyone has to handle, and no extra round trip is paid for the numbers. The alternative, a separate endpoint, buys only the ability to report statistics without registering — which is exactly the state worth not having. Rather than grow a third overload on each layer, the reporting arguments fold into the ignoreIfExists overloads that #8707 added and nothing yet calls. Catalog keeps the long-standing two-argument createPartitions plus the reporting one, and RESTApi is back to a single method; CachingCatalog, DelegateCatalog and RESTCatalog each lose a forwarder. No released signature changes. On FormatTablePartitionManager the reporting overload is the one an implementation provides and the two-argument form defaults onto it, rather than the other way round. An implementation that overrode only the short form would otherwise drop every report and leave the caller no way to notice, which is a rule the compiler can keep and a javadoc line cannot. Its twenty-five test doubles each spell the two arguments they ignore; that is the price, and the production implementation loses a forwarder in exchange. While here, supportsPartitionModification says in its javadoc that createPartitions and alterPartitions do nothing when it is false, which stopped being true once a REST catalog started registering the partitions of a Format Table while still reporting false. What the flag actually gates is the PartitionModification handler a table is given, so committing a table maintains no partitions; calling the catalog methods directly is unaffected. The javadoc now says that. The flag says how a report combines with what the server holds, per field: replacing overwrites all four values, adding sums the three counts and keeps the later creation time, since two timestamps do not add. Which one a writer sends follows from what it accounted for: an appending commit (INSERT INTO) adds, since it knows the files it wrote and not the ones the partition already held, and one statement whose writers commit separately reports each writer's share; an overwriting commit (INSERT OVERWRITE, the partition pinned statically or left to the data) replaces, since it left the partition holding what it wrote and nothing else. Reporting a field as unknown leaves the stored value alone either way, so a reporter that can measure only some fields never erases the others: a report here is read on the observation plane of PartitionStatistics, where a negative field was never measured, and not on the delta plane where it would be a decrement to subtract. Statistics never create or remove a partition row, whatever the numbers say. Compatibility runs both ways and neither direction errors. A new client against an old server: the fields are ignored, the statistics are dropped and the columns stay unknown. The client is not told, because the response carries no acknowledgement — an observation that is missing is a thing consumers can handle, an observation that is wrong is not, and an acknowledgement would be a second contract to keep in step with the first. An old client against a new server: no statistics arrive and the columns stay unknown. The two-argument createPartitions keeps sending exactly the request it sends today, so a caller that reports nothing does not change the shape the server sees. A non-empty adding report declares itself unsafe to retry, so an automatically replayed POST cannot count the same increment twice. A replacing report lands on the same value twice and keeps its retry, as does a request that reports nothing at all. Tests: HttpClientRetrySafetyTest covers which requests declare themselves unsafe to retry — only a non-empty adding report does — and that the flag itself never reaches the wire. RESTApiJsonTest covers the request shape, including that a client reporting nothing sends neither field. MockRESTCatalogTest covers reporting against the mock server end to end: adding accumulating, replacing overwriting, unknown fields leaving the stored value alone, statistics for a partition the request does not register being dropped, and a report that only partly matches not being applied at all. CatalogFormatTablePartitionManagerTest covers what the manager sends: statistics riding in the request of their own partitions, a batch that reports nothing sending an empty list rather than null, and the rejections of malformed reports. CachingCatalogTest and DelegateCatalogTest cover the new overload forwarding through and invalidating the partition cache. --- docs/static/rest-catalog-open-api.yaml | 8 + .../java/org/apache/paimon/rest/RESTApi.java | 41 +++- .../requests/CreatePartitionsRequest.java | 66 +++++- .../rest/HttpClientRetrySafetyTest.java | 109 +++++++--- .../apache/paimon/catalog/CachingCatalog.java | 9 +- .../org/apache/paimon/catalog/Catalog.java | 41 +++- .../paimon/catalog/DelegateCatalog.java | 9 +- .../org/apache/paimon/rest/RESTCatalog.java | 11 +- .../CatalogFormatTablePartitionManager.java | 103 ++++++++- .../table/format/FormatTableCommit.java | 4 +- .../format/FormatTablePartitionManager.java | 28 ++- .../paimon/catalog/CachingCatalogTest.java | 26 ++- .../paimon/catalog/DelegateCatalogTest.java | 87 ++++++++ .../paimon/rest/MockRESTCatalogTest.java | 160 ++++++++++++++ .../apache/paimon/rest/RESTApiJsonTest.java | 49 +++++ .../apache/paimon/rest/RESTCatalogServer.java | 114 +++++++++- .../apache/paimon/rest/RESTCatalogTest.java | 5 +- ...atalogFormatTablePartitionManagerTest.java | 202 +++++++++++++++++- .../CatalogManagedPartitionScanTest.java | 6 +- .../FormatTablePartitionRepairTest.java | 7 +- .../FormatTablePartitionDdlPlanningTest.scala | 41 ++-- .../FormatTablePartitionManagementTest.scala | 50 +++-- ...atalogManagedPartitionMsckRepairTest.scala | 13 +- 23 files changed, 1070 insertions(+), 119 deletions(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java diff --git a/docs/static/rest-catalog-open-api.yaml b/docs/static/rest-catalog-open-api.yaml index 452919f53f22..8550851f5029 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 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: 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 401921c8d17d..238a3969ff04 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 void markDonePartitions(Identifier identifier, List> restAuthFunction); } - /** Create partitions for table, ignoring partitions which already exist. */ - public CreatePartitionsResponse createPartitions( - Identifier identifier, List> 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. + * + *

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> partitions, boolean ignoreIfExists) { - CreatePartitionsRequest request = new CreatePartitionsRequest(partitions, ignoreIfExists); + Identifier identifier, + List> partitions, + boolean ignoreIfExists, + @Nullable List 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 506c66406eec..dadf1f9faa62 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 java.util.List; import java.util.Map; -/** Request for creating partitions. */ +/** + * Request for creating partitions. + * + *

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> 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; + + @JsonProperty(FIELD_REPLACE_STATISTICS) + @JsonInclude(JsonInclude.Include.NON_NULL) + @Nullable + private final Boolean replaceStatistics; + public CreatePartitionsRequest(List> partitionSpecs) { this(partitionSpecs, true); } + public CreatePartitionsRequest( + List> partitionSpecs, @Nullable Boolean ignoreIfExists) { + this(partitionSpecs, ignoreIfExists, null, null); + } + @JsonCreator public CreatePartitionsRequest( @JsonProperty(FIELD_PARTITION_SPECS) List> partitionSpecs, - @JsonProperty(FIELD_IGNORE_IF_EXISTS) @Nullable Boolean ignoreIfExists) { + @JsonProperty(FIELD_IGNORE_IF_EXISTS) @Nullable Boolean ignoreIfExists, + @JsonProperty(FIELD_PARTITION_STATISTICS) @Nullable + List 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> getPartitionSpecs() { public boolean ignoreIfExists() { return ignoreIfExists; } + + /** Reported statistics, or null when the client reports none. */ + @JsonGetter(FIELD_PARTITION_STATISTICS) + @Nullable + public List 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 8aca4d994776..d6c836dfcc3f 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.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. * *

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 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> 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 statistics) { + return new CreatePartitionsRequest( + Collections.singletonList(SPEC), true, statistics, replaceStatistics); + } + + private static List 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 01f04cbba3ae..e2a2558dc0d7 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 void createPartitions(Identifier identifier, List> pa @Override public void createPartitions( - Identifier identifier, List> partitions, boolean ignoreIfExists) + Identifier identifier, + List> partitions, + boolean ignoreIfExists, + @Nullable List 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 5b437efbab84..9653c67d01f8 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.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 @@ void deleteTag(Identifier identifier, String tagName) // ==================== 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. * - *

If not, following methods will do nothing: + *

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: * *

    *
  • {@link #createPartitions(Identifier, List)}. *
  • {@link #alterPartitions(Identifier, List)}. - *
- * - *

If not, following method will be exactly the same as directly using {@link - * BatchTableCommit#truncatePartitions}: - * - *

    *
  • {@link #dropPartitions(Identifier, List)}. *
+ * + *

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 @@ default void createPartitions(Identifier identifier, List> p 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. + * + *

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> partitions, boolean ignoreIfExists) + Identifier identifier, + List> partitions, + boolean ignoreIfExists, + @Nullable List 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 13faef105c11..6c691e6cee28 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 void createPartitions(Identifier identifier, List> pa @Override public void createPartitions( - Identifier identifier, List> partitions, boolean ignoreIfExists) + Identifier identifier, + List> partitions, + boolean ignoreIfExists, + @Nullable List 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 f1b5d94cc1a5..a24fb2354912 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 void markDonePartitions(Identifier identifier, List> @Override public void createPartitions(Identifier identifier, List> partitions) throws TableNotExistException { - createPartitions(identifier, partitions, true); + createPartitions(identifier, partitions, true, null, false); } @Override public void createPartitions( - Identifier identifier, List> partitions, boolean ignoreIfExists) + Identifier identifier, + List> partitions, + boolean ignoreIfExists, + @Nullable List 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 4d1009f196a4..1f4d4575a4ff 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.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 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 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 @@ private List collectAllPages(Map prefix, PageSupplier do { PagedList 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 @@ public List listPartitionsByNames(List> partition } @Override - public void createPartitions(List> partitions, boolean ignoreIfExists) { + public void createPartitions( + List> partitions, + boolean ignoreIfExists, + @Nullable List statistics, + boolean replaceStatistics) { + // Validated before the empty check: returning early would swallow a malformed report. + Map, PartitionStatistics> statisticsBySpec = + validateAndIndexStatistics(statistics, partitions); if (partitions.isEmpty()) { return; } @@ -163,19 +171,89 @@ public void createPartitions(List> partitions, boolean ignor 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> 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. + * + *

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, PartitionStatistics> validateAndIndexStatistics( + @Nullable List statistics, List> 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> registered = capacityFor(partitions.size()); + for (Map 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, 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 statisticsOf( + List> batch, + @Nullable Map, PartitionStatistics> statisticsBySpec) { + if (statisticsBySpec == null) { + return null; + } + List ofBatch = new ArrayList<>(batch.size()); + for (Map spec : batch) { + PartitionStatistics statistic = statisticsBySpec.get(spec); + if (statistic != null) { + ofBatch.add(statistic); + } + } + return ofBatch; + } + @Override public void dropPartitions(List> partitions) { if (partitions.isEmpty()) { @@ -193,6 +271,15 @@ public void dropPartitions(List> partitions) { "drop partitions"); } + private static Set 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 prefix) { for (Map.Entry 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 278e4c4f369f..152baa3ba010 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 void commit(List commitMessages) { 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 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 259f88bbfbf5..0e2c5c938533 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.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 listPartitionsByNames(List> 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> partitions, boolean ignoreIfExists); + default void createPartitions(List> 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. + * + *

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. + * + *

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> partitions, + boolean ignoreIfExists, + @Nullable List statistics, + boolean replaceStatistics); /** Unregister partitions. Metadata only; missing partitions are ignored. */ void dropPartitions(List> 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 0961050f110d..37b991fc6846 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.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 @@ public void testCreatePartitionsWithIgnoreIfExistsInvalidatesPartitionCache() th 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 spec = singletonMap("dt", "20260717"); + Partition created = new Partition(spec, 3, 300, 1, 1000, -1, false); + List 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 000000000000..ec230c8504bb --- /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> specs = + Arrays.asList( + Collections.singletonMap("dt", "20260728"), + Collections.singletonMap("dt", "20260729")); + List 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> 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 a68b7933e520..093c8f7bf5cc 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.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 @@ void testPartitionManagerSurvivesSerialization() throws Exception { .containsExactly(partition); } + @Test + void testReportedPartitionStatisticsAreStoredAndReadBack() throws Exception { + Identifier identifier = createFormatTableWithCatalogManagedPartitions(); + Map spec = Collections.singletonMap("dt", "20260717"); + List> 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 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 stored = Collections.singletonMap("dt", "20260717"); + Map 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 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 4a8b4ed96427..44449fcabcec 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 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 void createPartitionsRequestParseTest() throws Exception { 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 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 c571b19c3891..733b4b1c9fef 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 @@ private MockResponse partitionsApiHandle( List> existed = new ArrayList<>(); for (Map 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. + * + *

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 storedPartitions, + @Nullable List statistics, + @Nullable Boolean replaceStatistics) { + if (statistics == null) { + return; + } + boolean accumulate = !Boolean.TRUE.equals(replaceStatistics); + Map, PartitionStatistics> reported = new HashMap<>(); + for (PartitionStatistics statistic : statistics) { + reported.put(statistic.spec(), statistic); + } + Set> 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 @@ private synchronized MockResponse commitSnapshot( } 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 8fc5e38a25e3..87116e3e4576 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 @@ void testCreatePartitionsForCatalogManagedFormatTablePartitions() throws Excepti List> 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 @@ void testCreatePartitionsForCatalogManagedFormatTablePartitions() throws Excepti () -> 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 dfcc422be3b1..0c558973b7f3 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.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 @@ void testEmptyInputTouchesNoCatalog() { verifyNoInteractions(listCatalog); } + // ------------------------------------------------------------------------ + // reported statistics + // ------------------------------------------------------------------------ + + @Test + void testStatisticsRideInTheRequestOfTheirOwnPartitions() throws Exception { + Catalog catalog = mock(Catalog.class); + List> specs = specs(2500); + // Astride both split points, where a partition and its statistics could come apart. + List 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>> specCaptor = ArgumentCaptor.forClass(List.class); + @SuppressWarnings("unchecked") + ArgumentCaptor> statisticsCaptor = + ArgumentCaptor.forClass(List.class); + verify(catalog, times(3)) + .createPartitions( + eq(IDENTIFIER), + specCaptor.capture(), + eq(true), + statisticsCaptor.capture(), + eq(false)); + List>> requestedSpecs = specCaptor.getAllValues(); + List> 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> 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> specs = specs(2500); + // All in the first request, so the two that follow are the ones that report nothing. + List statistics = + Arrays.asList(statistics(specs.get(0), 1L), statistics(specs.get(999), 2L)); + + partitionManager(catalog).createPartitions(specs, true, statistics, false); + + @SuppressWarnings("unchecked") + ArgumentCaptor> statisticsCaptor = + ArgumentCaptor.forClass(List.class); + verify(catalog, times(3)) + .createPartitions( + eq(IDENTIFIER), anyList(), eq(true), statisticsCaptor.capture(), eq(false)); + List> 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> specs = specs(2500); + List 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> specs = Collections.singletonList(spec("2025", "01")); + // A spec typo would otherwise account for nothing at all, silently. + List 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 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 repeated = spec("2025", "01"); + List> specs = + Arrays.asList(repeated, spec("2025", "02"), spec("2025", "01")); + List 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 spec = spec("2025", "01"); + List 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> 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 @@ void testCheckedExceptionIsWrappedWithTableContext() throws Exception { 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 @@ private static FormatTablePartitionManager partitionManager(Catalog catalog) { return FormatTablePartitionManager.create(IDENTIFIER, PARTITION_KEYS, () -> catalog); } + /** The specs of every request a create that reports nothing sent. */ private static List>> capturedCreates( Catalog catalog, boolean ignoreIfExists, int expectedRequests) throws Exception { @SuppressWarnings("unchecked") ArgumentCaptor>> 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 @@ private static Map spec(String year, String month) { return spec; } + /** Statistics of one partition, told apart by their record count. */ + private static PartitionStatistics statistics(Map spec, long recordCount) { + return new PartitionStatistics( + spec, + recordCount, + recordCount * 1024, + 1L, + 1753660800000L, + PartitionStatistics.UNKNOWN_TOTAL_BUCKETS); + } + private static List> specs(int count) { List> 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 176dbf829145..56b7a184935f 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.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 @@ public List listPartitionsByNames(List> partition @Override public void createPartitions( - List> partitions, boolean ignoreIfExists) { + List> partitions, + boolean ignoreIfExists, + @Nullable List 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 bc6e24d219b7..033285c46e93 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.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 @@ private void register(List> partitions) { } @Override - public void createPartitions(List> partitions, boolean ignoreIfExists) { + public void createPartitions( + List> partitions, + boolean ignoreIfExists, + @Nullable List 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 7968c8d49cc6..8c24e43247a8 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 7c36fd09be0a..8febee0e0a0b 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 24e42b9c8685..586099b73b38 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 }