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

- * - *

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

+ * + *

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 }