From 0baf3572d8e7283972875206577bb0a590dc90a7 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 22 Jul 2026 03:37:50 +0800 Subject: [PATCH 1/4] [rest] Introduce partitions list-by-filter api definition The request carries the partition predicate in its filter field as the JSON serialization of a Paimon Predicate tree. The server evaluates it in a best-effort way: a tree it cannot restore or re-anchor counts as always-true, so the response is a superset of the matching partitions and never misses one, and callers keep applying the predicate on the returned partitions. Pages may be sparse; only an empty nextPageToken ends the listing. The mock server records received filter requests so tests can assert the transport path. --- .../java/org/apache/paimon/rest/RESTApi.java | 43 ++++++++ .../org/apache/paimon/rest/ResourcePaths.java | 12 +++ .../ListPartitionsByFilterRequest.java | 91 ++++++++++++++++ .../org/apache/paimon/catalog/Catalog.java | 29 +++++ .../paimon/catalog/DelegateCatalog.java | 13 +++ .../org/apache/paimon/rest/RESTCatalog.java | 28 +++++ .../apache/paimon/rest/RESTCatalogServer.java | 100 +++++++++++++++++- .../apache/paimon/rest/RESTCatalogTest.java | 91 ++++++++++++++++ 8 files changed, 406 insertions(+), 1 deletion(-) create mode 100644 paimon-api/src/main/java/org/apache/paimon/rest/requests/ListPartitionsByFilterRequest.java 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 4d321ed2797e..bd6e13236109 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 @@ -48,6 +48,7 @@ import org.apache.paimon.rest.requests.CreateViewRequest; import org.apache.paimon.rest.requests.DropPartitionsRequest; import org.apache.paimon.rest.requests.ForwardBranchRequest; +import org.apache.paimon.rest.requests.ListPartitionsByFilterRequest; import org.apache.paimon.rest.requests.ListPartitionsByNamesRequest; import org.apache.paimon.rest.requests.MarkDonePartitionsRequest; import org.apache.paimon.rest.requests.RegisterTableRequest; @@ -983,6 +984,48 @@ public List listPartitionsByNames( return partitions; } + /** + * List a page of partitions using a serialized partition predicate. + * + *

{@code filterJson} is the JSON serialization of a Paimon {@code Predicate}. The result may + * be a superset, so callers must re-evaluate the predicate. A non-empty next page token must be + * followed even when the current page is empty. + * + * @param identifier database name and table name + * @param filterJson JSON serialization of the partition predicate + * @param maxResults maximum page size, or {@code null}/0 to use the server default + * @param pageToken token returned by the previous page, or {@code null} for the first page + * @param partitionNamePattern optional SQL LIKE prefix pattern (%) for partition names, + * conjunctive with the predicate + * @return {@link PagedList}: elements and nextPageToken + * @throws NoSuchResourceException Exception thrown on HTTP 404 means the table not exists, or + * the server does not provide this endpoint + * @throws ForbiddenException Exception thrown on HTTP 403 means don't have the permission for + * this table + */ + public PagedList listPartitionsByFilterPaged( + Identifier identifier, + String filterJson, + @Nullable Integer maxResults, + @Nullable String pageToken, + @Nullable String partitionNamePattern) { + ListPartitionsByFilterRequest request = + new ListPartitionsByFilterRequest( + filterJson, partitionNamePattern, maxResults, pageToken); + ListPartitionsResponse response = + client.post( + resourcePaths.listPartitionsByFilter( + identifier.getDatabaseName(), identifier.getObjectName()), + request, + ListPartitionsResponse.class, + restAuthFunction); + List partitions = response.getPartitions(); + if (partitions == null) { + return new PagedList<>(emptyList(), null); + } + return new PagedList<>(partitions, response.getNextPageToken()); + } + /** * Create branch for table. * diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java index 0ab36e348c6e..ae09705467cc 100644 --- a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java +++ b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java @@ -249,6 +249,18 @@ public String listPartitionsByNames(String databaseName, String objectName) { "list-by-names"); } + public String listPartitionsByFilter(String databaseName, String objectName) { + return SLASH.join( + V1, + prefix, + DATABASES, + encodeString(databaseName), + TABLES, + encodeString(objectName), + PARTITIONS, + "list-by-filter"); + } + public String branches(String databaseName, String objectName) { return SLASH.join( V1, diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/requests/ListPartitionsByFilterRequest.java b/paimon-api/src/main/java/org/apache/paimon/rest/requests/ListPartitionsByFilterRequest.java new file mode 100644 index 000000000000..cd0fa058c4cd --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/rest/requests/ListPartitionsByFilterRequest.java @@ -0,0 +1,91 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.rest.requests; + +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.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; + +/** Request for listing partitions by filter. */ +@JsonIgnoreProperties(ignoreUnknown = true) +@JsonInclude(JsonInclude.Include.NON_NULL) +public class ListPartitionsByFilterRequest implements RESTRequest { + + private static final String FIELD_FILTER = "filter"; + private static final String FIELD_PARTITION_NAME_PATTERN = "partitionNamePattern"; + private static final String FIELD_MAX_RESULTS = "maxResults"; + private static final String FIELD_PAGE_TOKEN = "pageToken"; + + /** JSON serialization of a Paimon {@code Predicate} tree over the partition columns. */ + @JsonProperty(FIELD_FILTER) + private final String filter; + + @JsonProperty(FIELD_PARTITION_NAME_PATTERN) + @Nullable + private final String partitionNamePattern; + + @JsonProperty(FIELD_MAX_RESULTS) + @Nullable + private final Integer maxResults; + + @JsonProperty(FIELD_PAGE_TOKEN) + @Nullable + private final String pageToken; + + @JsonCreator + public ListPartitionsByFilterRequest( + @JsonProperty(FIELD_FILTER) String filter, + @JsonProperty(FIELD_PARTITION_NAME_PATTERN) @Nullable String partitionNamePattern, + @JsonProperty(FIELD_MAX_RESULTS) @Nullable Integer maxResults, + @JsonProperty(FIELD_PAGE_TOKEN) @Nullable String pageToken) { + this.filter = filter; + this.partitionNamePattern = partitionNamePattern; + this.maxResults = maxResults; + this.pageToken = pageToken; + } + + @JsonGetter(FIELD_FILTER) + public String getFilter() { + return filter; + } + + @JsonGetter(FIELD_PARTITION_NAME_PATTERN) + @Nullable + public String getPartitionNamePattern() { + return partitionNamePattern; + } + + @JsonGetter(FIELD_MAX_RESULTS) + @Nullable + public Integer getMaxResults() { + return maxResults; + } + + @JsonGetter(FIELD_PAGE_TOKEN) + @Nullable + public String getPageToken() { + return pageToken; + } +} 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 8e4ad8d51815..93f68bdf2e40 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 @@ -27,6 +27,7 @@ import org.apache.paimon.function.FunctionChange; import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.predicate.Predicate; import org.apache.paimon.rest.responses.GetTagResponse; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; @@ -423,6 +424,34 @@ PagedList listPartitionsPaged( @Nullable String partitionNamePattern) throws TableNotExistException; + /** + * Get a page of partitions using {@code predicate} as a best-effort filter. + * + *

The result may be a superset — an implementation must never exclude a partition it cannot + * prove non-matching — so callers must evaluate {@code predicate} again. Continue pagination + * while {@link PagedList#getNextPageToken()} is non-empty, even if a page has no elements. + * + * @param identifier path of the table + * @param predicate partition predicate + * @param maxResults maximum page size, or {@code null}/0 to use the catalog default + * @param pageToken token returned by the previous page, or {@code null} for the first page + * @param partitionNamePattern optional SQL LIKE prefix pattern (%) for partition names, + * conjunctive with {@code predicate} + * @return a page of candidate partitions + * @throws TableNotExistException if the table does not exist + * @throws UnsupportedOperationException if filtered partition listing is not supported + */ + default PagedList listPartitionsByFilterPaged( + Identifier identifier, + Predicate predicate, + @Nullable Integer maxResults, + @Nullable String pageToken, + @Nullable String partitionNamePattern) + throws TableNotExistException { + throw new UnsupportedOperationException( + this.getClass().getName() + " does not support listing partitions by filter."); + } + /** * Get Partition list by partition names of the table. * 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 290849024fdb..10ae54ea4594 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 @@ -24,6 +24,7 @@ import org.apache.paimon.function.FunctionChange; import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.predicate.Predicate; import org.apache.paimon.rest.responses.GetTagResponse; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; @@ -455,6 +456,18 @@ public PagedList listPartitionsPaged( return wrapped.listPartitionsPaged(identifier, maxResults, pageToken, partitionNamePattern); } + @Override + public PagedList listPartitionsByFilterPaged( + Identifier identifier, + Predicate predicate, + Integer maxResults, + String pageToken, + String partitionNamePattern) + throws TableNotExistException { + return wrapped.listPartitionsByFilterPaged( + identifier, predicate, maxResults, pageToken, partitionNamePattern); + } + @Override public List listPartitionsByNames( Identifier identifier, List> partitions) 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 efef63fc9275..86256a93425f 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 @@ -42,6 +42,7 @@ import org.apache.paimon.options.Options; import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.predicate.Predicate; import org.apache.paimon.rest.exceptions.AlreadyExistsException; import org.apache.paimon.rest.exceptions.BadRequestException; import org.apache.paimon.rest.exceptions.ForbiddenException; @@ -65,6 +66,7 @@ import org.apache.paimon.table.TableSnapshot; import org.apache.paimon.table.sink.BatchTableCommit; import org.apache.paimon.table.system.SystemTableLoader; +import org.apache.paimon.utils.JsonSerdeUtil; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.SnapshotNotExistException; import org.apache.paimon.utils.StringUtils; @@ -830,6 +832,32 @@ public PagedList listPartitionsPaged( } } + @Override + public PagedList listPartitionsByFilterPaged( + Identifier identifier, + Predicate predicate, + @Nullable Integer maxResults, + @Nullable String pageToken, + @Nullable String partitionNamePattern) + throws TableNotExistException { + try { + return api.listPartitionsByFilterPaged( + identifier, + JsonSerdeUtil.toFlatJson(predicate), + maxResults, + pageToken, + partitionNamePattern); + } catch (NoSuchResourceException e) { + throw new TableNotExistException(identifier); + } catch (ForbiddenException e) { + throw new TableNoPermissionException(identifier, e); + } catch (NotImplementedException e) { + // HTTP 501 is transport detail; callers only see the Catalog-level capability. + throw new UnsupportedOperationException( + "The REST server does not support listing partitions by filter.", e); + } + } + @Override public List listPartitionsByNames( Identifier identifier, List> partitions) 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 3a0e1539eea1..93401a0a81ee 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 @@ -28,8 +28,10 @@ import org.apache.paimon.catalog.PropertyChange; import org.apache.paimon.catalog.RenamingSnapshotCommit; import org.apache.paimon.catalog.TableMetadata; +import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.consumer.ConsumerInfo; import org.apache.paimon.consumer.ConsumerManager; +import org.apache.paimon.data.GenericRow; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; @@ -61,6 +63,7 @@ import org.apache.paimon.rest.requests.CreateTagRequest; import org.apache.paimon.rest.requests.CreateViewRequest; import org.apache.paimon.rest.requests.DropPartitionsRequest; +import org.apache.paimon.rest.requests.ListPartitionsByFilterRequest; import org.apache.paimon.rest.requests.ListPartitionsByNamesRequest; import org.apache.paimon.rest.requests.MarkDonePartitionsRequest; import org.apache.paimon.rest.requests.RenameTableRequest; @@ -109,8 +112,11 @@ import org.apache.paimon.table.TableSnapshot; import org.apache.paimon.table.object.ObjectTable; import org.apache.paimon.tag.Tag; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; import org.apache.paimon.utils.BranchManager; import org.apache.paimon.utils.ChangelogManager; +import org.apache.paimon.utils.InternalRowPartitionComputer; import org.apache.paimon.utils.JsonSerdeUtil; import org.apache.paimon.utils.LazyField; import org.apache.paimon.utils.Pair; @@ -147,6 +153,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -193,6 +200,11 @@ public class RESTCatalogServer { private final Map databaseStore = new HashMap<>(); private final Map tableMetadataStore = new HashMap<>(); + + /** Received filtered-listing requests, so tests can assert the transport path was taken. */ + public final List receivedListPartitionsByFilterRequests = + new java.util.concurrent.CopyOnWriteArrayList<>(); + private final Map> tablePartitionsStore = new HashMap<>(); private final Map viewStore = new HashMap<>(); private final Map tableLatestSnapshotStore = new HashMap<>(); @@ -459,6 +471,11 @@ && isTableByIdRequest(request.getPath())) { && ResourcePaths.TABLES.equals(resources[1]) && "partitions".equals(resources[3]) && "list-by-names".equals(resources[4]); + boolean isListPartitionsByFilter = + resources.length == 5 + && ResourcePaths.TABLES.equals(resources[1]) + && "partitions".equals(resources[3]) + && "list-by-filter".equals(resources[4]); boolean isDropPartitions = resources.length == 5 && ResourcePaths.TABLES.equals(resources[1]) @@ -511,7 +528,8 @@ && isTableByIdRequest(request.getPath())) { return new MockResponse().setResponseCode(200); } else if (!partitionListingSupported && ((isPartitions && "GET".equals(restAuthParameter.method())) - || isListPartitionsByNames)) { + || isListPartitionsByNames + || isListPartitionsByFilter)) { return mockResponse(new ErrorResponse(null, null, "", 501), 501); } else if (isDropPartitions) { return dropPartitionsHandle(restAuthParameter.data(), identifier); @@ -521,6 +539,11 @@ && isTableByIdRequest(request.getPath())) { restAuthParameter.data(), parameters, identifier); + } else if (isListPartitionsByFilter) { + ListPartitionsByFilterRequest listPartitionsByFilterRequest = + RESTApi.fromJson(data, ListPartitionsByFilterRequest.class); + return listPartitionsByFilter( + identifier, listPartitionsByFilterRequest); } else if (isListPartitionsByNames) { ListPartitionsByNamesRequest listPartitionsByNamesRequest = RESTApi.fromJson(data, ListPartitionsByNamesRequest.class); @@ -1943,6 +1966,81 @@ private MockResponse dropPartitionsHandle(String data, Identifier tableIdentifie return mockResponse(new DropPartitionsResponse(dropped, missing), 200); } + private MockResponse listPartitionsByFilter( + Identifier tableIdentifier, ListPartitionsByFilterRequest request) { + receivedListPartitionsByFilterRequests.add(request); + if (request.getFilter() == null || request.getFilter().isEmpty()) { + return mockResponse(new ErrorResponse(null, null, "filter is required", 400), 400); + } + TableMetadata metadata = tableMetadataStore.get(tableIdentifier.getFullName()); + RowType partitionType = metadata == null ? null : metadata.schema().logicalPartitionType(); + String defaultPartName = + metadata == null + ? CoreOptions.PARTITION_DEFAULT_NAME.defaultValue() + : new CoreOptions(metadata.schema().options()).partitionDefaultName(); + // Best-effort contract: a predicate the server cannot restore (version skew) or + // re-anchor (unknown column) counts as always-true, so the response stays a superset + // of the matching partitions and never misses one. + Predicate predicate; + try { + predicate = + TableQueryAuthResult.remapPredicate( + JsonSerdeUtil.fromJson(request.getFilter(), Predicate.class), + partitionType); + } catch (Exception e) { + predicate = null; + } + List partitions = new ArrayList<>(); + for (Map.Entry> entry : tablePartitionsStore.entrySet()) { + String objectName = Identifier.fromString(entry.getKey()).getObjectName(); + if (objectName.equals(tableIdentifier.getObjectName())) { + for (Partition partition : entry.getValue()) { + boolean patternMatched = + request.getPartitionNamePattern() == null + || matchNamePattern( + getPagedKey(partition), + request.getPartitionNamePattern()); + if (patternMatched + && matchesPredicate( + predicate, partition.spec(), partitionType, defaultPartName)) { + partitions.add(partition); + } + } + } + } + Map pagingParameters = new HashMap<>(); + if (request.getMaxResults() != null) { + pagingParameters.put(RESTApi.MAX_RESULTS, request.getMaxResults().toString()); + } + if (request.getPageToken() != null) { + pagingParameters.put(RESTApi.PAGE_TOKEN, request.getPageToken()); + } + return generateFinalListPartitionsResponse(pagingParameters, partitions); + } + + /** Evaluates the restored predicate against a spec; anything it cannot handle matches. */ + private static boolean matchesPredicate( + @Nullable Predicate predicate, + Map spec, + @Nullable RowType partitionType, + String defaultPartName) { + if (predicate == null || partitionType == null) { + return true; + } + try { + LinkedHashMap ordered = new LinkedHashMap<>(); + for (DataField field : partitionType.getFields()) { + ordered.put(field.name(), spec.get(field.name())); + } + GenericRow row = + InternalRowPartitionComputer.convertSpecToInternalRow( + ordered, partitionType, defaultPartName); + return predicate.test(row); + } catch (Exception e) { + return true; + } + } + private MockResponse listPartitionsByNames( Map parameters, Identifier tableIdentifier, 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 0e386e8b10cb..7e32eaf8ce66 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 @@ -96,6 +96,7 @@ import org.apache.paimon.table.system.SystemTableLoader; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; import org.apache.paimon.utils.SnapshotManager; import org.apache.paimon.utils.SnapshotNotExistException; import org.apache.paimon.utils.StringUtils; @@ -1877,6 +1878,96 @@ public void testListPartitionsPaged() throws Exception { () -> catalog.listPartitionsPaged(identifier, null, null, "dt=01%01")); } + @Test + public void testListPartitionsByFilterPaged() throws Exception { + if (!supportPartitions()) { + return; + } + + String databaseName = "partitions_filter_db"; + List> partitionSpecs = + Arrays.asList( + singletonMap("dt", "20250101"), + singletonMap("dt", "20250102"), + singletonMap("dt", "20250103"), + singletonMap("dt", "20260101")); + catalog.dropDatabase(databaseName, true, true); + catalog.createDatabase(databaseName, true); + Identifier identifier = Identifier.create(databaseName, "table"); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .option(METASTORE_TAG_TO_PARTITION.key(), "dt") + .column("col", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .build(), + true); + BatchWriteBuilder writeBuilder = catalog.getTable(identifier).newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + for (Map partitionSpec : partitionSpecs) { + write.write(GenericRow.of(0, BinaryString.fromString(partitionSpec.get("dt")))); + } + commit.commit(write.prepareCommit()); + } + + // The predicate goes on the wire as its own JSON serialization; the server evaluates + // the very same tree the scan would evaluate locally. + PredicateBuilder builder = + new PredicateBuilder( + RowType.of( + new org.apache.paimon.types.DataType[] {DataTypes.STRING()}, + new String[] {"dt"})); + Predicate range = + PredicateBuilder.and( + builder.greaterOrEqual(0, BinaryString.fromString("20250102")), + builder.lessThan(0, BinaryString.fromString("20260101"))); + assertThat( + catalog.listPartitionsByFilterPaged(identifier, range, null, null, null) + .getElements()) + .extracting(partition -> partition.spec().get("dt")) + .containsExactlyInAnyOrder("20250102", "20250103"); + + Predicate in = + builder.in( + 0, + Arrays.asList( + BinaryString.fromString("20250101"), + BinaryString.fromString("20260101"))); + assertThat( + catalog.listPartitionsByFilterPaged(identifier, in, null, null, null) + .getElements()) + .extracting(partition -> partition.spec().get("dt")) + .containsExactlyInAnyOrder("20250101", "20260101"); + + // A pattern combines with the predicate as a conjunction. + assertThat( + catalog.listPartitionsByFilterPaged(identifier, in, null, null, "dt=2025%") + .getElements()) + .extracting(partition -> partition.spec().get("dt")) + .containsExactly("20250101"); + + // A predicate the server cannot re-anchor (unknown column) counts as always-true: the + // response is a superset of the matching partitions and the client keeps filtering + // locally. + PredicateBuilder unknownColumn = + new PredicateBuilder( + RowType.of( + new org.apache.paimon.types.DataType[] {DataTypes.STRING()}, + new String[] {"nope"})); + assertThat( + catalog.listPartitionsByFilterPaged( + identifier, + unknownColumn.equal(0, BinaryString.fromString("x")), + null, + null, + null) + .getElements()) + .hasSize(4); + } + @Test public void testListPartitionsPagedWithMultiLevel() throws Exception { if (!supportPartitions()) { From 9b577f531a4ad14de61d15724d1c3186af254f52 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 22 Jul 2026 03:37:50 +0800 Subject: [PATCH 2/4] [core] Push partition filters to the REST catalog for Format Table scans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Send the partition predicate of a managed Format Table scan to the partitions/list-by-filter endpoint together with the existing name prefix. The predicate travels as its own flat JSON serialization, so the server evaluates exactly the tree the scan evaluates locally and there is no translation step; the scan keeps no pushdown policy of its own — whatever partition predicate exists goes to the manager. Request sizing is left to the HTTP layer, and the server weakens what it cannot evaluate instead of rejecting it, returning a superset (today that includes temporal and decimal literals, whose JSON round trip stays unrestorable until the literal serde is hardened upstream). Falling back is stateless and stays inside the one catalog the operation loaded: a catalog without the endpoint (an UnsupportedOperationException, into which the REST catalog translates HTTP 501, or the endpoint 404) costs one rejected probe per filtered scan and the same call retries the plain prefix listing, so behavior is identical across copies, serialization and processes. The per-partition filter in the plan is unchanged, and a Spark suite pins the chain end to end: a Spark SQL partition predicate must reach the filtered listing endpoint and restore to the pushed tree. --- .../CatalogFormatTablePartitionManager.java | 102 ++++++-- .../format/FormatTablePartitionManager.java | 10 +- .../paimon/table/format/FormatTableScan.java | 20 +- .../paimon/rest/MockRESTCatalogTest.java | 8 +- ...atalogFormatTablePartitionManagerTest.java | 220 +++++++++++++++++- .../CatalogManagedPartitionScanTest.java | 26 ++- .../format/FormatTablePartitionRepair.java | 3 +- .../spark/format/PaimonFormatTable.scala | 2 +- .../FormatTablePartitionRepairTest.java | 9 +- .../PaimonSparkTestWithRestCatalogBase.scala | 2 +- .../FormatTablePartitionDdlPlanningTest.scala | 38 ++- .../FormatTablePartitionManagementTest.scala | 45 +++- ...atalogManagedPartitionMsckRepairTest.scala | 7 +- ...nagedFormatTablePartitionPruningTest.scala | 80 +++++++ 14 files changed, 494 insertions(+), 78 deletions(-) create mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ManagedFormatTablePartitionPruningTest.scala 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 7085135f17ff..a5a160c4ae6d 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,10 +23,13 @@ import org.apache.paimon.catalog.CatalogLoader; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.partition.Partition; +import org.apache.paimon.predicate.Predicate; import org.apache.paimon.utils.FunctionWithException; import org.apache.paimon.utils.PartitionPathUtils; import org.apache.paimon.utils.StringUtils; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; @@ -62,7 +65,7 @@ class CatalogFormatTablePartitionManager implements FormatTablePartitionManager } @Override - public List listPartitions(Map prefix) { + public List listPartitions(Map prefix, @Nullable Predicate filter) { LinkedHashMap ordered = orderedPrefix(partitionKeys, prefix); checkArgument( ordered.size() == prefix.size(), @@ -73,31 +76,88 @@ public List listPartitions(Map prefix) { String pattern = PartitionPathUtils.buildPartitionNamePrefixPattern(partitionKeys, ordered); return execute( catalog -> { - List partitions = new ArrayList<>(); - Set seenPageTokens = new HashSet<>(); - String pageToken = null; - do { - PagedList page = - catalog.listPartitionsPaged( - identifier, REQUEST_SIZE, pageToken, pattern); - if (page.getElements() != null) { - partitions.addAll(page.getElements()); - } - pageToken = page.getNextPageToken(); - if (StringUtils.isNotEmpty(pageToken) && !seenPageTokens.add(pageToken)) { - throw new IllegalStateException( - String.format( - "Catalog returned repeated partition page token '%s' for format table %s.", - pageToken, identifier.getFullName())); + if (filter != null) { + try { + return collectAllPages( + prefix, + pageToken -> + catalog.listPartitionsByFilterPaged( + identifier, + filter, + REQUEST_SIZE, + pageToken, + pattern)); + } catch (Exception e) { + if (!indicatesNoFilterListing(e)) { + throw e; + } + // The filtered listing endpoint is optional: retry as the plain + // prefix listing, which returns a superset the caller filters + // anyway. A genuinely missing table also lands here (the + // endpoint's 404 cannot be told apart); the plain listing below + // then reports it properly. } - } while (StringUtils.isNotEmpty(pageToken)); - // A catalog may ignore the pattern, so the prefix is enforced here as well. - partitions.removeIf(partition -> !matchesPrefix(partition, prefix)); - return Collections.unmodifiableList(partitions); + } + return collectAllPages( + prefix, + pageToken -> + catalog.listPartitionsPaged( + identifier, REQUEST_SIZE, pageToken, pattern)); }, "list partitions"); } + /** + * Drains a paged listing. Pages may be sparse (fewer elements than requested, or none, with a + * next page token), so only an empty token ends the loop. + */ + private List collectAllPages(Map prefix, PageSupplier pageSupplier) + throws Exception { + List partitions = new ArrayList<>(); + Set seenPageTokens = new HashSet<>(); + String pageToken = null; + 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. + throw new IllegalStateException( + String.format( + "Catalog returned a null partition page for format table %s.", + identifier.getFullName())); + } + if (page.getElements() != null) { + partitions.addAll(page.getElements()); + } + pageToken = page.getNextPageToken(); + if (StringUtils.isNotEmpty(pageToken) && !seenPageTokens.add(pageToken)) { + throw new IllegalStateException( + String.format( + "Catalog returned repeated partition page token '%s' for format table %s.", + pageToken, identifier.getFullName())); + } + } while (StringUtils.isNotEmpty(pageToken)); + // A catalog may ignore the pattern, so the prefix is enforced here as well. + partitions.removeIf(partition -> !matchesPrefix(partition, prefix)); + return Collections.unmodifiableList(partitions); + } + + /** A single page of a partition listing. */ + private interface PageSupplier { + PagedList page(@Nullable String pageToken) throws Exception; + } + + /** + * Whether the failure means the catalog cannot serve a filtered partition listing. {@link + * Catalog.TableNotExistException} doubles as the endpoint-missing signal — a REST catalog + * cannot tell an endpoint 404 from a missing table — and the plain retry reports a genuinely + * missing table properly. + */ + private static boolean indicatesNoFilterListing(Exception exception) { + return exception instanceof UnsupportedOperationException + || exception instanceof Catalog.TableNotExistException; + } + @Override public List listPartitionsByNames(List> partitions) { if (partitions.isEmpty()) { 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 64b643fba26e..259f88bbfbf5 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,9 @@ import org.apache.paimon.catalog.CatalogLoader; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.partition.Partition; +import org.apache.paimon.predicate.Predicate; + +import javax.annotation.Nullable; import java.io.Serializable; import java.util.List; @@ -46,8 +49,13 @@ public interface FormatTablePartitionManager extends Serializable { * List every partition whose leading partition values match {@code prefix}, or all partitions * when it is empty. The prefix must be a contiguous leading subset of the table's partition * keys. + * + *

The optional {@code filter} is a pushdown hint combined with the prefix as a conjunction: + * an implementation may apply it partially or not at all, so the result is a superset of the + * matching partitions and never misses one. Callers keep applying their full predicate on the + * returned partitions. */ - List listPartitions(Map prefix); + List listPartitions(Map prefix, @Nullable Predicate filter); /** Return those of the given complete partition specs that are registered. */ List listPartitionsByNames(List> partitions); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index 44908749cad6..2dee6fe1237b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -121,7 +121,8 @@ public Plan plan() { @Override public List listPartitionEntries() { if (partitionManager != null) { - List partitions = partitionManager.listPartitions(Collections.emptyMap()); + List partitions = + partitionManager.listPartitions(Collections.emptyMap(), null); if (partitions.isEmpty()) { warnIfFilesystemPartitionsExist(); } @@ -228,13 +229,17 @@ public List splits() { List, Path>> findPartitions() { if (partitionManager != null) { - Map prefix = leadingEqualityPrefix(); - List partitions = partitionManager.listPartitions(prefix); - if (partitions.isEmpty() && prefix.isEmpty()) { + Optional extracted = extractPartitionPredicate(partitionFilter); + Map prefix = leadingEqualityPrefix(extracted); + // The whole predicate goes to the manager as a pushdown hint, together with the + // prefix; the catalog may return a superset. + Predicate catalogFilter = extracted.orElse(null); + List partitions = partitionManager.listPartitions(prefix, catalogFilter); + if (partitions.isEmpty() && prefix.isEmpty() && catalogFilter == null) { warnIfFilesystemPartitionsExist(); } - // The prefix is a coarse pre-filter; the full predicate is applied per partition in - // the plan, as it is for filesystem discovery. + // The prefix and filter are coarse pre-filters; the full predicate is applied per + // partition in the plan, as it is for filesystem discovery. return toSpecsAndPaths(partitions, coreOptions.formatTablePartitionOnlyValueInPath()); } LOG.debug( @@ -342,8 +347,7 @@ private IllegalStateException corruptPartitionSpec(@Nullable Map * to the partition catalog. Empty when there is no predicate or it does not start with * equalities on the leading partition keys. */ - private Map leadingEqualityPrefix() { - Optional predicate = extractPartitionPredicate(partitionFilter); + private Map leadingEqualityPrefix(Optional predicate) { if (!predicate.isPresent()) { return Collections.emptyMap(); } 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 6cc8f7042018..b942a82a503a 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 @@ -258,19 +258,19 @@ void testCatalogManagedPartitionListingReflectsCatalogMutationsImmediately() thr FormatTable table = (FormatTable) restCatalog.getTable(identifier); FormatTablePartitionManager partitionManager = table.partitionManager(); assertThat(partitionManager).isNotNull(); - assertThat(partitionManager.listPartitions(Collections.emptyMap())).isEmpty(); + assertThat(partitionManager.listPartitions(Collections.emptyMap(), null)).isEmpty(); Map partition = Collections.singletonMap("dt", "20260717"); restCatalog.createPartitions(identifier, Collections.singletonList(partition)); // Listings are not cached, so a mutation through the catalog is visible to the next read. - assertThat(partitionManager.listPartitions(Collections.emptyMap())) + assertThat(partitionManager.listPartitions(Collections.emptyMap(), null)) .extracting(org.apache.paimon.partition.Partition::spec) .containsExactly(partition); restCatalog.dropPartitions(identifier, Collections.singletonList(partition)); - assertThat(partitionManager.listPartitions(Collections.emptyMap())).isEmpty(); + assertThat(partitionManager.listPartitions(Collections.emptyMap(), null)).isEmpty(); } @Test @@ -284,7 +284,7 @@ void testPartitionManagerSurvivesSerialization() throws Exception { FormatTablePartitionManager roundTripped = InstantiationUtil.clone(table.partitionManager()); - assertThat(roundTripped.listPartitions(Collections.emptyMap())) + assertThat(roundTripped.listPartitions(Collections.emptyMap(), null)) .extracting(org.apache.paimon.partition.Partition::spec) .containsExactly(partition); } 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 d7a5d97e691a..006b94d938fd 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 @@ -22,7 +22,13 @@ import org.apache.paimon.catalog.Catalog; import org.apache.paimon.catalog.CatalogLoader; import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryString; import org.apache.paimon.partition.Partition; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; import org.apache.paimon.utils.InstantiationUtil; import org.junit.jupiter.api.Test; @@ -46,6 +52,7 @@ import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; @@ -66,6 +73,14 @@ class CatalogFormatTablePartitionManagerTest { /** One catalog request never carries more than this many partitions. */ private static final int REQUEST_SIZE = 1000; + /** Any non-null partition predicate; the manager passes it through untouched. */ + private static final Predicate FILTER = + new PredicateBuilder( + RowType.of( + new DataType[] {DataTypes.STRING(), DataTypes.STRING()}, + new String[] {"year", "month"})) + .greaterThan(1, BinaryString.fromString("01")); + /** Handed to the serializable loader, which cannot capture a test instance field. */ private static Catalog staticCatalog; @@ -84,7 +99,7 @@ void testListPartitionsFollowsEveryPage() throws Exception { new PagedList<>(Collections.singletonList(second), null)); List partitions = - partitionManager(catalog).listPartitions(Collections.emptyMap()); + partitionManager(catalog).listPartitions(Collections.emptyMap(), null); assertThat(partitions).containsExactly(first, second); ArgumentCaptor pageTokens = ArgumentCaptor.forClass(String.class); @@ -101,7 +116,7 @@ void testEmptyNextPageTokenEndsPaging() throws Exception { .thenReturn(new PagedList<>(Collections.singletonList(only), "")); List partitions = - partitionManager(catalog).listPartitions(Collections.emptyMap()); + partitionManager(catalog).listPartitions(Collections.emptyMap(), null); assertThat(partitions).containsExactly(only); verify(catalog, times(1)).listPartitionsPaged(any(), any(), any(), any()); @@ -119,12 +134,192 @@ void testRepeatedPageTokenFailsFast() throws Exception { FormatTablePartitionManager partitionManager = partitionManager(catalog); - assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap())) + assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap(), null)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("repeated partition page token") .hasMessageContaining("catalog_partition_db.catalog_partition_table"); } + // ------------------------------------------------------------------------ + // filtered listing and its fallback + // ------------------------------------------------------------------------ + + @Test + void testFilteredListingPassesFilterPrefixAndTokens() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition first = partition("2025", "01"); + Partition second = partition("2025", "02"); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenReturn( + new PagedList<>(Collections.singletonList(first), "page-2"), + new PagedList<>(Collections.singletonList(second), null)); + + List partitions = + partitionManager(catalog) + .listPartitions(Collections.singletonMap("year", "2025"), FILTER); + + assertThat(partitions).containsExactly(first, second); + ArgumentCaptor pageTokens = ArgumentCaptor.forClass(String.class); + verify(catalog, times(2)) + .listPartitionsByFilterPaged( + eq(IDENTIFIER), + eq(FILTER), + eq(REQUEST_SIZE), + pageTokens.capture(), + eq("year=2025/%")); + assertThat(pageTokens.getAllValues()).containsExactly(null, "page-2"); + verify(catalog, never()).listPartitionsPaged(any(), any(), any(), any()); + } + + @Test + void testFilteredListingFollowsSparsePages() throws Exception { + // Filtering may leave a page empty while more partitions remain to check: only an empty + // token ends the loop. + Catalog catalog = mock(Catalog.class); + Partition only = partition("2025", "02"); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenReturn( + new PagedList<>(Collections.emptyList(), "page-2"), + new PagedList<>(Collections.singletonList(only), null)); + + List partitions = + partitionManager(catalog).listPartitions(Collections.emptyMap(), FILTER); + + assertThat(partitions).containsExactly(only); + verify(catalog, times(2)).listPartitionsByFilterPaged(any(), any(), any(), any(), any()); + } + + @Test + void testFilteredListingRepeatedTokenFailsFast() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenReturn( + new PagedList<>(Collections.singletonList(partition("2025", "01")), "a"), + new PagedList<>(Collections.singletonList(partition("2025", "02")), "a")); + FormatTablePartitionManager partitionManager = partitionManager(catalog); + + assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap(), FILTER)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("repeated partition page token"); + } + + @Test + void testFilteredPrefixIsEnforcedOnCatalogsThatIgnoreThePattern() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenReturn( + new PagedList<>( + Arrays.asList(partition("2024", "12"), partition("2025", "01")), + null)); + + List partitions = + partitionManager(catalog) + .listPartitions(Collections.singletonMap("year", "2025"), FILTER); + + assertThat(partitions).containsExactly(partition("2025", "01")); + } + + @Test + void testFallbackOnCatalogWithoutFilteredListing() throws Exception { + // UnsupportedOperationException is the Catalog-level capability signal (the Catalog + // default method, or a REST catalog translating HTTP 501). + Catalog catalog = mock(Catalog.class); + Partition only = partition("2025", "01"); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenThrow(new UnsupportedOperationException("no filtered listing")); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Collections.singletonList(only), null)); + + List partitions = + partitionManager(catalog) + .listPartitions(Collections.singletonMap("year", "2025"), FILTER); + + assertThat(partitions).containsExactly(only); + verify(catalog).listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), any(), any()); + } + + @Test + void testFallbackOnEndpointMissing() throws Exception { + // An old REST server without the route reports the endpoint's 404 as + // TableNotExistException; the plain retry serves the listing. + Catalog catalog = mock(Catalog.class); + Partition only = partition("2025", "01"); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenThrow(new Catalog.TableNotExistException(IDENTIFIER)); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Collections.singletonList(only), null)); + + List partitions = + partitionManager(catalog).listPartitions(Collections.emptyMap(), FILTER); + + assertThat(partitions).containsExactly(only); + } + + @Test + void testFallbackUsesTheSameCatalog() throws Exception { + // One high-level operation loads one catalog: the fallback retries on the catalog the + // filtered attempt used instead of loading a second one. + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenThrow(new UnsupportedOperationException("no filtered listing")); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Collections.emptyList(), null)); + int[] loads = new int[1]; + FormatTablePartitionManager partitionManager = + FormatTablePartitionManager.create( + IDENTIFIER, + PARTITION_KEYS, + () -> { + loads[0]++; + return catalog; + }); + + partitionManager.listPartitions(Collections.emptyMap(), FILTER); + + assertThat(loads[0]).isEqualTo(1); + verify(catalog, times(1)).close(); + } + + @Test + void testOtherFilteredFailuresDoNotFallBack() throws Exception { + // Permission problems, server failures and malformed requests must fail loudly; a + // silent plain retry would hide them. + Catalog catalog = mock(Catalog.class); + RuntimeException failure = new IllegalStateException("permission denied"); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenThrow(failure); + FormatTablePartitionManager partitionManager = partitionManager(catalog); + + Throwable thrown = + catchThrowable( + () -> partitionManager.listPartitions(Collections.emptyMap(), FILTER)); + + assertThat(thrown).isSameAs(failure); + verify(catalog, never()).listPartitionsPaged(any(), any(), any(), any()); + } + + @Test + void testMissingTableStillFailsAfterFallback() throws Exception { + // The endpoint's 404 and a missing table cannot be told apart, so the plain retry is + // the arbiter: a genuinely missing table fails there with full context. + Catalog catalog = mock(Catalog.class); + Catalog.TableNotExistException notExist = new Catalog.TableNotExistException(IDENTIFIER); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenThrow(new Catalog.TableNotExistException(IDENTIFIER)); + when(catalog.listPartitionsPaged(any(), any(), any(), any())).thenThrow(notExist); + FormatTablePartitionManager partitionManager = partitionManager(catalog); + + Throwable thrown = + catchThrowable( + () -> partitionManager.listPartitions(Collections.emptyMap(), FILTER)); + + assertThat(thrown) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("list partitions") + .hasMessageContaining("catalog_partition_db.catalog_partition_table"); + assertThat(thrown.getCause()).isSameAs(notExist); + } + // ------------------------------------------------------------------------ // prefix pushdown and client side filtering // ------------------------------------------------------------------------ @@ -135,7 +330,7 @@ void testLeadingPrefixIsPushedDownAsPattern() throws Exception { when(catalog.listPartitionsPaged(any(), any(), any(), any())) .thenReturn(new PagedList<>(Collections.emptyList(), null)); - partitionManager(catalog).listPartitions(Collections.singletonMap("year", "2025")); + partitionManager(catalog).listPartitions(Collections.singletonMap("year", "2025"), null); verify(catalog) .listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), isNull(), eq("year=2025/%")); @@ -147,7 +342,7 @@ void testEmptyPrefixPushesDownNoPattern() throws Exception { when(catalog.listPartitionsPaged(any(), any(), any(), any())) .thenReturn(new PagedList<>(Collections.emptyList(), null)); - partitionManager(catalog).listPartitions(Collections.emptyMap()); + partitionManager(catalog).listPartitions(Collections.emptyMap(), null); verify(catalog).listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), isNull(), isNull()); } @@ -161,7 +356,8 @@ void testPrefixIsEnforcedOnCatalogsThatIgnoreThePattern() throws Exception { .thenReturn(new PagedList<>(Arrays.asList(matching, other), null)); List partitions = - partitionManager(catalog).listPartitions(Collections.singletonMap("year", "2025")); + partitionManager(catalog) + .listPartitions(Collections.singletonMap("year", "2025"), null); assertThat(partitions).containsExactly(matching); } @@ -173,7 +369,7 @@ void testNonLeadingPrefixIsRejected() { // "month" without "year" cannot be expressed as a partition path prefix. Map prefix = Collections.singletonMap("month", "10"); - assertThatThrownBy(() -> partitionManager.listPartitions(prefix)) + assertThatThrownBy(() -> partitionManager.listPartitions(prefix, null)) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("leading prefix"); } @@ -268,7 +464,7 @@ void testEveryOperationClosesTheCatalog() throws Exception { Catalog listCatalog = mock(Catalog.class); when(listCatalog.listPartitionsPaged(any(), any(), any(), any())) .thenReturn(new PagedList<>(Collections.emptyList(), null)); - partitionManager(listCatalog).listPartitions(Collections.emptyMap()); + partitionManager(listCatalog).listPartitions(Collections.emptyMap(), null); verify(listCatalog).close(); Catalog byNamesCatalog = mock(Catalog.class); @@ -293,7 +489,7 @@ void testFailingOperationStillClosesTheCatalog() throws Exception { .thenThrow(new RuntimeException("catalog unavailable")); FormatTablePartitionManager partitionManager = partitionManager(catalog); - assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap())) + assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap(), null)) .isInstanceOf(RuntimeException.class); verify(catalog).close(); @@ -305,7 +501,7 @@ void testRejectedPrefixNeverLoadsACatalog() { FormatTablePartitionManager partitionManager = partitionManager(catalog); Map prefix = Collections.singletonMap("month", "10"); - assertThatThrownBy(() -> partitionManager.listPartitions(prefix)) + assertThatThrownBy(() -> partitionManager.listPartitions(prefix, null)) .isInstanceOf(IllegalArgumentException.class); verifyNoInteractions(catalog); @@ -323,7 +519,7 @@ void testCheckedExceptionIsWrappedWithTableContext() throws Exception { FormatTablePartitionManager partitionManager = partitionManager(catalog); Throwable thrown = - catchThrowable(() -> partitionManager.listPartitions(Collections.emptyMap())); + catchThrowable(() -> partitionManager.listPartitions(Collections.emptyMap(), null)); assertThat(thrown) .isInstanceOf(RuntimeException.class) @@ -360,7 +556,7 @@ void testSurvivesJavaSerialization() throws Exception { FormatTablePartitionManager.create(IDENTIFIER, PARTITION_KEYS, new TestLoader()); FormatTablePartitionManager cloned = InstantiationUtil.clone(partitionManager); - assertThat(cloned.listPartitions(Collections.emptyMap())).containsExactly(only); + assertThat(cloned.listPartitions(Collections.emptyMap(), null)).containsExactly(only); verify(catalog).close(); } 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 da74ea32732f..aa4c9c7a7ea3 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 @@ -41,6 +41,8 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import javax.annotation.Nullable; + import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; @@ -54,9 +56,11 @@ import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** Tests for Format Table scans whose partitions are catalog-managed. */ @@ -75,6 +79,18 @@ void testLeadingPatternResidualFilterAndUnregisteredDirectory() throws Exception new PagedList<>( Arrays.asList(partition("2025", "10"), partition("2025", "11")), null)); + // The residual predicate beyond the prefix goes to the filter endpoint; the filter is a + // hint, so returning a superset here is legal and the plan still filters per partition. + when(catalog.listPartitionsByFilterPaged( + eq(IDENTIFIER), + any(Predicate.class), + eq(1000), + isNull(), + eq("year=2025/%"))) + .thenReturn( + new PagedList<>( + Arrays.asList(partition("2025", "10"), partition("2025", "11")), + null)); TrackingLocalFileIO fileIO = new TrackingLocalFileIO(); Path tablePath = new Path(tempDir.toUri()); Path octoberFile = writeDataFile(fileIO, tablePath, "year=2025/month=10"); @@ -93,6 +109,13 @@ void testLeadingPatternResidualFilterAndUnregisteredDirectory() throws Exception assertThat(plannedFiles).containsExactly(novemberFile); assertThat(plannedFiles).doesNotContain(octoberFile, unregisteredFile); assertThat(fileIO.listedPaths).containsExactly(new Path(tablePath, "year=2025/month=11")); + verify(catalog) + .listPartitionsByFilterPaged( + eq(IDENTIFIER), + any(Predicate.class), + eq(1000), + isNull(), + eq("year=2025/%")); } @Test @@ -313,7 +336,8 @@ private FormatTablePartitionManager recordingCatalog(List partitions) List> prefixes = requestedPrefixes; return new FormatTablePartitionManager() { @Override - public List listPartitions(Map prefix) { + public List listPartitions( + Map prefix, @Nullable Predicate filter) { prefixes.add(new LinkedHashMap<>(prefix)); List matching = new ArrayList<>(); for (Partition partition : partitions) { diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java index 4aa99b812c14..d77968f248dd 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java @@ -109,7 +109,8 @@ static int apply( boolean addPartitions, boolean dropPartitions) { Set> registeredPartitions = new HashSet<>(); - for (Partition partition : partitionManager.listPartitions(Collections.emptyMap())) { + for (Partition partition : + partitionManager.listPartitions(Collections.emptyMap(), null)) { registeredPartitions.add(partition.spec()); } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala index 776fc98ca1e4..2138a0cb1495 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala @@ -164,7 +164,7 @@ case class PaimonFormatTable(table: FormatTable) // One unfiltered traversal resolves every partial spec; the requested constraints are // enforced client-side. requirePartitionManager() - .listPartitions(Collections.emptyMap[String, String]()) + .listPartitions(Collections.emptyMap[String, String](), null) .asScala .foreach { partition => 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 5bf21ded1873..bc6e24d219b7 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.predicate.Predicate; import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.format.FormatTablePartitionManager; import org.apache.paimon.types.DataTypes; @@ -33,6 +34,8 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import javax.annotation.Nullable; + import java.io.IOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; @@ -307,7 +310,8 @@ void repairFailsClosedWhenTheCatalogListingFails() throws Exception { RecordingPartitionManager catalog = new RecordingPartitionManager() { @Override - public List listPartitions(Map prefix) { + public List listPartitions( + Map prefix, @Nullable Predicate filter) { listCount.incrementAndGet(); throw listFailure; } @@ -456,7 +460,8 @@ public List listPartitionsByNames(List> partition } @Override - public List listPartitions(Map prefix) { + public List listPartitions( + Map prefix, @Nullable Predicate filter) { requestedPrefixes.add(prefix); List partitions = new ArrayList<>(registered.size()); for (Map spec : registered) { diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala index c880ba7f3657..0c0035373e03 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala @@ -32,7 +32,7 @@ import java.util.UUID class PaimonSparkTestWithRestCatalogBase extends PaimonSparkTestBase { - private var restCatalogServer: RESTCatalogServer = _ + protected var restCatalogServer: RESTCatalogServer = _ private var serverUrl: String = _ protected var warehouse: String = _ private val initToken = "init_token" 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 7ef72caad0dc..7968c8d49cc6 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 @@ -23,6 +23,7 @@ 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.predicate.Predicate import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase import org.apache.paimon.spark.catalyst.plans.logical.PaimonDropPartitions import org.apache.paimon.spark.format.PaimonFormatTable @@ -375,7 +376,9 @@ class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalog partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = throw new AssertionError("Complete specs must resolve through list-by-names") } @@ -429,7 +432,9 @@ class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalog partitions.asScala.map(_.asScala.toMap).filter(_ == registeredSpec).toSeq: _*) } - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = throw new AssertionError("Complete specs must resolve through list-by-names") } @@ -495,7 +500,9 @@ class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalog partitions: JList[JMap[String, String]]): JList[Partition] = registeredPartitions(partitions.asScala.map(_.asScala.toMap).filter(registered).toSeq: _*) - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = throw new AssertionError("Exact specs must resolve through list-by-names") } var refreshCalls = 0 @@ -562,7 +569,9 @@ class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalog partitions: JList[JMap[String, String]]): JList[Partition] = registeredPartitions(partitions.asScala.map(_.asScala.toMap).filter(registered).toSeq: _*) - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = throw new AssertionError("Exact specs must resolve through list-by-names") } val firstFileIO = LocalFileIO.create @@ -616,7 +625,9 @@ class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalog partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = { + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = { listedPrefixes :+= prefix.asScala.toMap registeredPartitions(matching) } @@ -656,7 +667,9 @@ class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalog partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = { + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = { listedPrefixes :+= prefix.asScala.toMap registeredPartitions(first, unrelated, second, third) } @@ -708,7 +721,9 @@ class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalog partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = registeredPartitions(Map("dt" -> "20260715")) } val table = new PaimonFormatTable(rawFormatTable(withCatalogManagedPartitions = true, gateway)) @@ -912,7 +927,7 @@ class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalog registeredPartitions(partitions.asScala.map(_.asScala.toMap).toSeq: _*) } - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions(prefix: JMap[String, String], filter: Predicate): JList[Partition] = Collections.emptyList() } @@ -944,8 +959,9 @@ class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalog override def listPartitionsByNames(partitions: JList[JMap[String, String]]): JList[Partition] = throw new AssertionError("ADD must not perform a client-side existence lookup") - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = synchronized { - registeredPartitions(partitions.toSeq: _*) - } + override def listPartitions(prefix: JMap[String, String], filter: Predicate): JList[Partition] = + synchronized { + registeredPartitions(partitions.toSeq: _*) + } } } 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 3cb44485decc..7c36fd09be0a 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 @@ -23,6 +23,7 @@ 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.predicate.Predicate import org.apache.paimon.table.FormatTable import org.apache.paimon.table.format.FormatTablePartitionManager import org.apache.paimon.types.DataTypes @@ -61,7 +62,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = Collections.emptyList() } val sparkTable = @@ -113,7 +116,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = Collections.emptyList() } @@ -144,7 +149,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = Collections.emptyList() } @@ -174,7 +181,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = Collections.emptyList() } @@ -245,7 +254,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = Collections.emptyList() } @@ -293,7 +304,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = { + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = { listedPrefixes :+= prefix.asScala.toMap registeredPartitions(first, second) } @@ -407,7 +420,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = registeredPartitions(matching) } @@ -446,7 +461,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = throw new TestPartitionListException } @@ -500,7 +517,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = registeredPartitions(registered.toSeq: _*) } @@ -600,7 +619,9 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions( + prefix: JMap[String, String], + filter: Predicate): JList[Partition] = Collections.emptyList() } @@ -639,7 +660,7 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { partitions.asScala.map(_.asScala.toMap).filter(storedPartitions.contains).toSeq: _*) } - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions(prefix: JMap[String, String], filter: Predicate): JList[Partition] = throw new AssertionError("ADD must not enumerate catalog or filesystem partitions") } @@ -662,7 +683,7 @@ class FormatTablePartitionManagementTest extends SparkFunSuite { override def listPartitionsByNames(partitions: JList[JMap[String, String]]): JList[Partition] = Collections.emptyList() - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = + override def listPartitions(prefix: JMap[String, String], filter: Predicate): JList[Partition] = Collections.emptyList() } 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 3997341b455d..24e42b9c8685 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 @@ -21,6 +21,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.predicate.Predicate import org.apache.paimon.spark.{PaimonSparkTestWithRestCatalogBase, SparkCatalog} import org.apache.paimon.spark.execution.PaimonRepairFormatTablePartitionsExec import org.apache.paimon.spark.format.PaimonFormatTable @@ -579,7 +580,7 @@ private[sql] class StatefulFaultCatalog(initial: Set[Map[String, String]] = Set. MsckTestFixtures.toPartitions( partitions.asScala.map(_.asScala.toMap).filter(state.contains).toSeq) - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = { + override def listPartitions(prefix: JMap[String, String], filter: Predicate): JList[Partition] = { if (failList) { throw new IllegalStateException(MsckFaultInjection.LIST_FAILURE) } @@ -659,11 +660,11 @@ private[sql] class FaultInjectingFormatTablePartitionManager(delegate: FormatTab override def listPartitionsByNames(partitions: JList[JMap[String, String]]): JList[Partition] = delegate.listPartitionsByNames(partitions) - override def listPartitions(prefix: JMap[String, String]): JList[Partition] = { + override def listPartitions(prefix: JMap[String, String], filter: Predicate): JList[Partition] = { MsckFaultInjection.listCalls += 1 if (MsckFaultInjection.failList) { throw new IllegalStateException(MsckFaultInjection.LIST_FAILURE) } - delegate.listPartitions(prefix) + delegate.listPartitions(prefix, filter) } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ManagedFormatTablePartitionPruningTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ManagedFormatTablePartitionPruningTest.scala new file mode 100644 index 000000000000..959336c6f0fe --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ManagedFormatTablePartitionPruningTest.scala @@ -0,0 +1,80 @@ +/* + * 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.spark.sql + +import org.apache.paimon.predicate.{LeafPredicate, Predicate, PredicateBuilder} +import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase +import org.apache.paimon.utils.JsonSerdeUtil + +import org.apache.spark.sql.Row + +import scala.collection.JavaConverters._ + +/** + * End-to-end guard of the pushdown chain: a Spark SQL partition predicate on a catalog-managed + * Format Table must reach the REST `partitions/list-by-filter` endpoint, not only produce correct + * rows (the client-side residual filter would mask a broken transport path). + */ +class ManagedFormatTablePartitionPruningTest extends PaimonSparkTestWithRestCatalogBase { + + test("Format Table: partition predicate reaches the filtered listing endpoint") { + val tableName = "pruning_endpoint_t" + withTable(tableName) { + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, year INT, month INT) + |USING CSV + |PARTITIONED BY (year, month) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + sql(s"""INSERT INTO $tableName VALUES + |(1, 'a', 2024, 12), (2, 'b', 2025, 9), (3, 'c', 2025, 11), (4, 'd', 2025, 12) + |""".stripMargin) + + restCatalogServer.receivedListPartitionsByFilterRequests.clear() + checkAnswer( + sql(s"SELECT id FROM $tableName WHERE year = 2025 AND month > 10 ORDER BY id"), + Seq(Row(3), Row(4))) + + val requests = restCatalogServer.receivedListPartitionsByFilterRequests + assert(!requests.isEmpty, "the filtered listing endpoint was not called") + val request = requests.get(requests.size() - 1) + assert(request.getFilter != null && request.getFilter.nonEmpty) + assert( + request.getPartitionNamePattern != null + && request.getPartitionNamePattern.startsWith("year=2025"), + s"unexpected pattern ${request.getPartitionNamePattern}" + ) + + // The filter restores to the pushed predicate (the wire form round-trips) and uses the + // flat encoding, no pretty-print newlines. + val restored = JsonSerdeUtil.fromJson(request.getFilter, classOf[Predicate]) + val leafFields = PredicateBuilder + .splitAnd(restored) + .asScala + .collect { case leaf: LeafPredicate => leaf.fieldNames().asScala } + .flatten + .toSet + assert( + leafFields.contains("year") && leafFields.contains("month"), + s"restored filter misses pushed fields: $restored") + assert(!request.getFilter.contains("\n"), "filter is not flat-encoded") + } + } +} From aaf9bb4517169f8b5d0a1286e649aeffd55a3b51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?= Date: Wed, 22 Jul 2026 10:46:00 +0800 Subject: [PATCH 3/4] [core][rest] Refine filtered partition listing and tests (#3) --- .../java/org/apache/paimon/rest/RESTApi.java | 2 +- .../org/apache/paimon/catalog/Catalog.java | 4 +- .../CatalogFormatTablePartitionManager.java | 48 +++----- .../paimon/rest/MockRESTCatalogTest.java | 44 +++++++ .../apache/paimon/rest/RESTApiJsonTest.java | 28 +++++ .../apache/paimon/rest/RESTCatalogServer.java | 57 ++++++--- .../apache/paimon/rest/RESTCatalogTest.java | 85 +++++++------ ...atalogFormatTablePartitionManagerTest.java | 113 +++--------------- .../CatalogManagedPartitionScanTest.java | 46 ++++++- .../PaimonSparkTestWithRestCatalogBase.scala | 10 +- ...cala => CatalogManagedPartitionTest.scala} | 26 +++- ...nagedFormatTablePartitionPruningTest.scala | 80 ------------- 12 files changed, 267 insertions(+), 276 deletions(-) rename paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/{CatalogManagedPartitionLoadTest.scala => CatalogManagedPartitionTest.scala} (62%) delete mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ManagedFormatTablePartitionPruningTest.scala 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 bd6e13236109..e8e1ff96c87b 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 @@ -1021,7 +1021,7 @@ public PagedList listPartitionsByFilterPaged( restAuthFunction); List partitions = response.getPartitions(); if (partitions == null) { - return new PagedList<>(emptyList(), null); + return new PagedList<>(emptyList(), response.getNextPageToken()); } return new PagedList<>(partitions, response.getNextPageToken()); } 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 93f68bdf2e40..fd477ffcf149 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 @@ -439,7 +439,6 @@ PagedList listPartitionsPaged( * conjunctive with {@code predicate} * @return a page of candidate partitions * @throws TableNotExistException if the table does not exist - * @throws UnsupportedOperationException if filtered partition listing is not supported */ default PagedList listPartitionsByFilterPaged( Identifier identifier, @@ -448,8 +447,7 @@ default PagedList listPartitionsByFilterPaged( @Nullable String pageToken, @Nullable String partitionNamePattern) throws TableNotExistException { - throw new UnsupportedOperationException( - this.getClass().getName() + " does not support listing partitions by filter."); + return listPartitionsPaged(identifier, maxResults, pageToken, partitionNamePattern); } /** 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 a5a160c4ae6d..ed266db2a546 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 @@ -77,26 +77,15 @@ public List listPartitions(Map prefix, @Nullable Pred return execute( catalog -> { if (filter != null) { - try { - return collectAllPages( - prefix, - pageToken -> - catalog.listPartitionsByFilterPaged( - identifier, - filter, - REQUEST_SIZE, - pageToken, - pattern)); - } catch (Exception e) { - if (!indicatesNoFilterListing(e)) { - throw e; - } - // The filtered listing endpoint is optional: retry as the plain - // prefix listing, which returns a superset the caller filters - // anyway. A genuinely missing table also lands here (the - // endpoint's 404 cannot be told apart); the plain listing below - // then reports it properly. - } + return collectAllPages( + prefix, + pageToken -> + catalog.listPartitionsByFilterPaged( + identifier, + filter, + REQUEST_SIZE, + pageToken, + pattern)); } return collectAllPages( prefix, @@ -127,7 +116,11 @@ private List collectAllPages(Map prefix, PageSupplier identifier.getFullName())); } if (page.getElements() != null) { - partitions.addAll(page.getElements()); + for (Partition partition : page.getElements()) { + if (matchesPrefix(partition, prefix)) { + partitions.add(partition); + } + } } pageToken = page.getNextPageToken(); if (StringUtils.isNotEmpty(pageToken) && !seenPageTokens.add(pageToken)) { @@ -137,8 +130,6 @@ private List collectAllPages(Map prefix, PageSupplier pageToken, identifier.getFullName())); } } while (StringUtils.isNotEmpty(pageToken)); - // A catalog may ignore the pattern, so the prefix is enforced here as well. - partitions.removeIf(partition -> !matchesPrefix(partition, prefix)); return Collections.unmodifiableList(partitions); } @@ -147,17 +138,6 @@ private interface PageSupplier { PagedList page(@Nullable String pageToken) throws Exception; } - /** - * Whether the failure means the catalog cannot serve a filtered partition listing. {@link - * Catalog.TableNotExistException} doubles as the endpoint-missing signal — a REST catalog - * cannot tell an endpoint 404 from a missing table — and the plain retry reports a genuinely - * missing table properly. - */ - private static boolean indicatesNoFilterListing(Exception exception) { - return exception instanceof UnsupportedOperationException - || exception instanceof Catalog.TableNotExistException; - } - @Override public List listPartitionsByNames(List> partitions) { if (partitions.isEmpty()) { 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 b942a82a503a..ff3736d88903 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 @@ -30,6 +30,7 @@ import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.options.CatalogOptions; import org.apache.paimon.options.Options; +import org.apache.paimon.partition.Partition; import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; @@ -63,6 +64,7 @@ import java.io.File; import java.io.IOException; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -289,6 +291,40 @@ void testPartitionManagerSurvivesSerialization() throws Exception { .containsExactly(partition); } + @Test + void testFilteredListingPreservesNextTokenAcrossSparsePage() throws Exception { + Identifier identifier = createFormatTableWithCatalogManagedPartitions(); + Predicate predicate = partitionFilter("20260717"); + Partition partition = + new Partition(Collections.singletonMap("dt", "20260717"), 0, 0, 0, 0, -1, false); + restCatalogServer.enqueueListPartitionsByFilterResponse(null, "p2"); + restCatalogServer.enqueueListPartitionsByFilterResponse( + Collections.singletonList(partition), null); + + PagedList firstPage = + restCatalog.listPartitionsByFilterPaged(identifier, predicate, 1, null, "dt=2026%"); + assertThat(firstPage.getElements()).isEmpty(); + assertThat(firstPage.getNextPageToken()).isEqualTo("p2"); + + PagedList secondPage = + restCatalog.listPartitionsByFilterPaged( + identifier, predicate, 1, firstPage.getNextPageToken(), "dt=2026%"); + assertThat(secondPage.getElements()).containsExactly(partition); + assertThat(secondPage.getNextPageToken()).isNull(); + + assertThat(restCatalogServer.getReceivedListPartitionsByFilterRequests()) + .extracting( + request -> + Arrays.asList( + request.getFilter(), + request.getMaxResults(), + request.getPageToken(), + request.getPartitionNamePattern())) + .containsExactly( + Arrays.asList(JsonSerdeUtil.toFlatJson(predicate), 1, null, "dt=2026%"), + Arrays.asList(JsonSerdeUtil.toFlatJson(predicate), 1, "p2", "dt=2026%")); + } + @Test void testRejectCatalogManagedPartitionsOnExternalTableBeforeCreate() throws Exception { Identifier identifier = Identifier.create("db1", "external_partitioned_format_table"); @@ -351,6 +387,14 @@ private Identifier createFormatTableWithCatalogManagedPartitions() throws Except return identifier; } + private static Predicate partitionFilter(String value) { + return new PredicateBuilder( + RowType.of( + new org.apache.paimon.types.DataType[] {DataTypes.STRING()}, + new String[] {"dt"})) + .equal(0, value); + } + @Test void testBaseHeadersInRequests() throws Exception { // Set custom headers in options 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 bdf2b5f82c1b..7508e004737f 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 @@ -28,6 +28,7 @@ import org.apache.paimon.rest.requests.CreateTableRequest; import org.apache.paimon.rest.requests.CreateViewRequest; import org.apache.paimon.rest.requests.DropPartitionsRequest; +import org.apache.paimon.rest.requests.ListPartitionsByFilterRequest; import org.apache.paimon.rest.requests.RenameTableRequest; import org.apache.paimon.rest.requests.RollbackTableRequest; import org.apache.paimon.rest.responses.AlterDatabaseResponse; @@ -210,6 +211,33 @@ public void listPartitionsResponseParseTest() throws Exception { parseData.getPartitions().get(0).fileCount()); } + @Test + public void listPartitionsByFilterRequestJsonShapeTest() throws Exception { + ListPartitionsByFilterRequest request = + new ListPartitionsByFilterRequest("filter-json", "dt=2026%", 2, "p2"); + String json = RESTApi.toJson(request); + + Map expected = new HashMap<>(); + expected.put("filter", "filter-json"); + expected.put("partitionNamePattern", "dt=2026%"); + expected.put("maxResults", 2); + expected.put("pageToken", "p2"); + assertEquals(expected, RESTApi.fromJson(json, Map.class)); + ListPartitionsByFilterRequest parsed = + RESTApi.fromJson(json, ListPartitionsByFilterRequest.class); + assertEquals(request.getFilter(), parsed.getFilter()); + assertEquals(request.getPartitionNamePattern(), parsed.getPartitionNamePattern()); + assertEquals(request.getMaxResults(), parsed.getMaxResults()); + assertEquals(request.getPageToken(), parsed.getPageToken()); + + assertEquals( + Collections.singletonMap("filter", "filter-json"), + RESTApi.fromJson( + RESTApi.toJson( + new ListPartitionsByFilterRequest("filter-json", null, null, null)), + Map.class)); + } + @Test public void createPartitionsResponseParseTest() throws Exception { Map created = new HashMap<>(); 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 93401a0a81ee..6796e1ddf79e 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 @@ -158,8 +158,10 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; +import java.util.Queue; import java.util.Set; import java.util.UUID; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.regex.Pattern; import java.util.stream.Collectors; @@ -201,10 +203,12 @@ public class RESTCatalogServer { private final Map databaseStore = new HashMap<>(); private final Map tableMetadataStore = new HashMap<>(); - /** Received filtered-listing requests, so tests can assert the transport path was taken. */ - public final List receivedListPartitionsByFilterRequests = + private final List receivedListPartitionsByFilterRequests = new java.util.concurrent.CopyOnWriteArrayList<>(); + private final Queue scriptedListPartitionsByFilterResponses = + new ConcurrentLinkedQueue<>(); + private final Map> tablePartitionsStore = new HashMap<>(); private final Map viewStore = new HashMap<>(); private final Map tableLatestSnapshotStore = new HashMap<>(); @@ -291,6 +295,25 @@ public void setPartitionListingSupported(boolean partitionListingSupported) { this.partitionListingSupported = partitionListingSupported; } + public void clearReceivedListPartitionsByFilterRequests() { + receivedListPartitionsByFilterRequests.clear(); + } + + public List getReceivedListPartitionsByFilterRequests() { + return Collections.unmodifiableList( + new ArrayList<>(receivedListPartitionsByFilterRequests)); + } + + public boolean hasReceivedListPartitionsByFilterRequest() { + return !receivedListPartitionsByFilterRequests.isEmpty(); + } + + public void enqueueListPartitionsByFilterResponse( + @Nullable List partitions, @Nullable String nextPageToken) { + scriptedListPartitionsByFilterResponses.add( + new ListPartitionsResponse(partitions, nextPageToken)); + } + public void addNoPermissionDatabase(String database) { noPermissionDatabases.add(database); } @@ -1969,6 +1992,10 @@ private MockResponse dropPartitionsHandle(String data, Identifier tableIdentifie private MockResponse listPartitionsByFilter( Identifier tableIdentifier, ListPartitionsByFilterRequest request) { receivedListPartitionsByFilterRequests.add(request); + ListPartitionsResponse scriptedResponse = scriptedListPartitionsByFilterResponses.poll(); + if (scriptedResponse != null) { + return mockResponse(scriptedResponse, 200); + } if (request.getFilter() == null || request.getFilter().isEmpty()) { return mockResponse(new ErrorResponse(null, null, "filter is required", 400), 400); } @@ -1991,21 +2018,17 @@ private MockResponse listPartitionsByFilter( predicate = null; } List partitions = new ArrayList<>(); - for (Map.Entry> entry : tablePartitionsStore.entrySet()) { - String objectName = Identifier.fromString(entry.getKey()).getObjectName(); - if (objectName.equals(tableIdentifier.getObjectName())) { - for (Partition partition : entry.getValue()) { - boolean patternMatched = - request.getPartitionNamePattern() == null - || matchNamePattern( - getPagedKey(partition), - request.getPartitionNamePattern()); - if (patternMatched - && matchesPredicate( - predicate, partition.spec(), partitionType, defaultPartName)) { - partitions.add(partition); - } - } + for (Partition partition : + tablePartitionsStore.getOrDefault( + tableIdentifier.getFullName(), Collections.emptyList())) { + boolean patternMatched = + request.getPartitionNamePattern() == null + || matchNamePattern( + getPagedKey(partition), request.getPartitionNamePattern()); + if (patternMatched + && matchesPredicate( + predicate, partition.spec(), partitionType, defaultPartName)) { + partitions.add(partition); } } Map pagingParameters = new HashMap<>(); 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 7e32eaf8ce66..05376d3585b5 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 @@ -373,6 +373,17 @@ void testApiWhenTableNoPermission() throws Exception { assertThrows( Catalog.TableNoPermissionException.class, () -> catalog.listPartitionsPaged(identifier, 100, null, null)); + Predicate partitionFilter = + new PredicateBuilder( + RowType.of( + new org.apache.paimon.types.DataType[] {DataTypes.INT()}, + new String[] {"col1"})) + .equal(0, 1); + assertThrows( + Catalog.TableNoPermissionException.class, + () -> + catalog.listPartitionsByFilterPaged( + identifier, partitionFilter, 100, null, null)); assertThrows( Catalog.TableNoPermissionException.class, () -> restCatalog.createBranch(identifier, "test_branch", null)); @@ -1891,19 +1902,30 @@ public void testListPartitionsByFilterPaged() throws Exception { singletonMap("dt", "20250102"), singletonMap("dt", "20250103"), singletonMap("dt", "20260101")); - catalog.dropDatabase(databaseName, true, true); - catalog.createDatabase(databaseName, true); - Identifier identifier = Identifier.create(databaseName, "table"); - catalog.createTable( - identifier, + Schema schema = Schema.newBuilder() .option(METASTORE_PARTITIONED_TABLE.key(), "true") .option(METASTORE_TAG_TO_PARTITION.key(), "dt") .column("col", DataTypes.INT()) .column("dt", DataTypes.STRING()) .partitionKeys("dt") - .build(), - true); + .build(); + PredicateBuilder builder = + new PredicateBuilder( + RowType.of( + new org.apache.paimon.types.DataType[] {DataTypes.STRING()}, + new String[] {"dt"})); + Predicate range = + PredicateBuilder.and( + builder.greaterOrEqual(0, BinaryString.fromString("20250101")), + builder.lessThan(0, BinaryString.fromString("20260101"))); + catalog.dropDatabase(databaseName, true, true); + catalog.createDatabase(databaseName, true); + Identifier identifier = Identifier.create(databaseName, "table"); + assertThrows( + Catalog.TableNotExistException.class, + () -> catalog.listPartitionsByFilterPaged(identifier, range, 2, null, "dt=2025%")); + catalog.createTable(identifier, schema, true); BatchWriteBuilder writeBuilder = catalog.getTable(identifier).newBatchWriteBuilder(); try (BatchTableWrite write = writeBuilder.newWrite(); BatchTableCommit commit = writeBuilder.newCommit()) { @@ -1913,41 +1935,29 @@ public void testListPartitionsByFilterPaged() throws Exception { commit.commit(write.prepareCommit()); } + String distractorDatabase = databaseName + "_other"; + catalog.dropDatabase(distractorDatabase, true, true); + catalog.createDatabase(distractorDatabase, true); + Identifier distractor = Identifier.create(distractorDatabase, identifier.getObjectName()); + catalog.createTable(distractor, schema, true); + catalog.createPartitions(distractor, singletonList(singletonMap("dt", "20250104"))); + // The predicate goes on the wire as its own JSON serialization; the server evaluates // the very same tree the scan would evaluate locally. - PredicateBuilder builder = - new PredicateBuilder( - RowType.of( - new org.apache.paimon.types.DataType[] {DataTypes.STRING()}, - new String[] {"dt"})); - Predicate range = - PredicateBuilder.and( - builder.greaterOrEqual(0, BinaryString.fromString("20250102")), - builder.lessThan(0, BinaryString.fromString("20260101"))); - assertThat( - catalog.listPartitionsByFilterPaged(identifier, range, null, null, null) - .getElements()) + PagedList firstPage = + catalog.listPartitionsByFilterPaged(identifier, range, 2, null, "dt=2025%"); + assertThat(firstPage.getElements()) .extracting(partition -> partition.spec().get("dt")) - .containsExactlyInAnyOrder("20250102", "20250103"); + .containsExactly("20250103", "20250102"); + assertThat(firstPage.getNextPageToken()).isEqualTo("dt=20250102"); - Predicate in = - builder.in( - 0, - Arrays.asList( - BinaryString.fromString("20250101"), - BinaryString.fromString("20260101"))); - assertThat( - catalog.listPartitionsByFilterPaged(identifier, in, null, null, null) - .getElements()) - .extracting(partition -> partition.spec().get("dt")) - .containsExactlyInAnyOrder("20250101", "20260101"); - - // A pattern combines with the predicate as a conjunction. - assertThat( - catalog.listPartitionsByFilterPaged(identifier, in, null, null, "dt=2025%") - .getElements()) + PagedList secondPage = + catalog.listPartitionsByFilterPaged( + identifier, range, 2, firstPage.getNextPageToken(), "dt=2025%"); + assertThat(secondPage.getElements()) .extracting(partition -> partition.spec().get("dt")) .containsExactly("20250101"); + assertThat(secondPage.getNextPageToken()).isNull(); // A predicate the server cannot re-anchor (unknown column) counts as always-true: the // response is a superset of the matching partitions and the client keeps filtering @@ -1965,7 +1975,8 @@ public void testListPartitionsByFilterPaged() throws Exception { null, null) .getElements()) - .hasSize(4); + .extracting(Partition::spec) + .containsExactlyInAnyOrderElementsOf(partitionSpecs); } @Test 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 006b94d938fd..7d246bf4f583 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 @@ -70,7 +70,6 @@ class CatalogFormatTablePartitionManagerTest { private static final List PARTITION_KEYS = Arrays.asList("year", "month"); - /** One catalog request never carries more than this many partitions. */ private static final int REQUEST_SIZE = 1000; /** Any non-null partition predicate; the manager passes it through untouched. */ @@ -130,7 +129,8 @@ void testRepeatedPageTokenFailsFast() throws Exception { .thenReturn( new PagedList<>(Collections.singletonList(partition("2025", "01")), "a"), new PagedList<>(Collections.singletonList(partition("2025", "02")), "b"), - new PagedList<>(Collections.singletonList(partition("2025", "03")), "a")); + new PagedList<>(Collections.singletonList(partition("2025", "03")), "a")) + .thenThrow(new AssertionError("unexpected extra page request")); FormatTablePartitionManager partitionManager = partitionManager(catalog); @@ -141,17 +141,17 @@ void testRepeatedPageTokenFailsFast() throws Exception { } // ------------------------------------------------------------------------ - // filtered listing and its fallback + // filtered listing // ------------------------------------------------------------------------ @Test - void testFilteredListingPassesFilterPrefixAndTokens() throws Exception { + void testFilteredListingPassesRequestAndPreservesSupersetAcrossPages() throws Exception { Catalog catalog = mock(Catalog.class); Partition first = partition("2025", "01"); Partition second = partition("2025", "02"); when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) .thenReturn( - new PagedList<>(Collections.singletonList(first), "page-2"), + new PagedList<>(Arrays.asList(partition("2024", "12"), first), "page-2"), new PagedList<>(Collections.singletonList(second), null)); List partitions = @@ -190,102 +190,29 @@ void testFilteredListingFollowsSparsePages() throws Exception { } @Test - void testFilteredListingRepeatedTokenFailsFast() throws Exception { - Catalog catalog = mock(Catalog.class); - when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) - .thenReturn( - new PagedList<>(Collections.singletonList(partition("2025", "01")), "a"), - new PagedList<>(Collections.singletonList(partition("2025", "02")), "a")); - FormatTablePartitionManager partitionManager = partitionManager(catalog); - - assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap(), FILTER)) - .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("repeated partition page token"); - } - - @Test - void testFilteredPrefixIsEnforcedOnCatalogsThatIgnoreThePattern() throws Exception { - Catalog catalog = mock(Catalog.class); - when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) - .thenReturn( - new PagedList<>( - Arrays.asList(partition("2024", "12"), partition("2025", "01")), - null)); - - List partitions = - partitionManager(catalog) - .listPartitions(Collections.singletonMap("year", "2025"), FILTER); - - assertThat(partitions).containsExactly(partition("2025", "01")); - } - - @Test - void testFallbackOnCatalogWithoutFilteredListing() throws Exception { - // UnsupportedOperationException is the Catalog-level capability signal (the Catalog - // default method, or a REST catalog translating HTTP 501). - Catalog catalog = mock(Catalog.class); - Partition only = partition("2025", "01"); - when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) - .thenThrow(new UnsupportedOperationException("no filtered listing")); - when(catalog.listPartitionsPaged(any(), any(), any(), any())) - .thenReturn(new PagedList<>(Collections.singletonList(only), null)); - - List partitions = - partitionManager(catalog) - .listPartitions(Collections.singletonMap("year", "2025"), FILTER); - - assertThat(partitions).containsExactly(only); - verify(catalog).listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), any(), any()); - } - - @Test - void testFallbackOnEndpointMissing() throws Exception { - // An old REST server without the route reports the endpoint's 404 as - // TableNotExistException; the plain retry serves the listing. + void testCatalogDefaultFilteredListingDelegatesToPlainListing() throws Exception { Catalog catalog = mock(Catalog.class); Partition only = partition("2025", "01"); when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) - .thenThrow(new Catalog.TableNotExistException(IDENTIFIER)); + .thenCallRealMethod(); when(catalog.listPartitionsPaged(any(), any(), any(), any())) .thenReturn(new PagedList<>(Collections.singletonList(only), null)); - List partitions = - partitionManager(catalog).listPartitions(Collections.emptyMap(), FILTER); + PagedList page = + catalog.listPartitionsByFilterPaged( + IDENTIFIER, FILTER, REQUEST_SIZE, "page-2", "year=2025/%"); - assertThat(partitions).containsExactly(only); - } - - @Test - void testFallbackUsesTheSameCatalog() throws Exception { - // One high-level operation loads one catalog: the fallback retries on the catalog the - // filtered attempt used instead of loading a second one. - Catalog catalog = mock(Catalog.class); - when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) - .thenThrow(new UnsupportedOperationException("no filtered listing")); - when(catalog.listPartitionsPaged(any(), any(), any(), any())) - .thenReturn(new PagedList<>(Collections.emptyList(), null)); - int[] loads = new int[1]; - FormatTablePartitionManager partitionManager = - FormatTablePartitionManager.create( - IDENTIFIER, - PARTITION_KEYS, - () -> { - loads[0]++; - return catalog; - }); - - partitionManager.listPartitions(Collections.emptyMap(), FILTER); - - assertThat(loads[0]).isEqualTo(1); - verify(catalog, times(1)).close(); + assertThat(page.getElements()).containsExactly(only); + verify(catalog) + .listPartitionsPaged( + eq(IDENTIFIER), eq(REQUEST_SIZE), eq("page-2"), eq("year=2025/%")); } @Test void testOtherFilteredFailuresDoNotFallBack() throws Exception { - // Permission problems, server failures and malformed requests must fail loudly; a - // silent plain retry would hide them. Catalog catalog = mock(Catalog.class); - RuntimeException failure = new IllegalStateException("permission denied"); + Catalog.TableNoPermissionException failure = + new Catalog.TableNoPermissionException(IDENTIFIER); when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) .thenThrow(failure); FormatTablePartitionManager partitionManager = partitionManager(catalog); @@ -299,14 +226,11 @@ void testOtherFilteredFailuresDoNotFallBack() throws Exception { } @Test - void testMissingTableStillFailsAfterFallback() throws Exception { - // The endpoint's 404 and a missing table cannot be told apart, so the plain retry is - // the arbiter: a genuinely missing table fails there with full context. + void testMissingTableDoesNotTriggerPlainListing() throws Exception { Catalog catalog = mock(Catalog.class); Catalog.TableNotExistException notExist = new Catalog.TableNotExistException(IDENTIFIER); when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) - .thenThrow(new Catalog.TableNotExistException(IDENTIFIER)); - when(catalog.listPartitionsPaged(any(), any(), any(), any())).thenThrow(notExist); + .thenThrow(notExist); FormatTablePartitionManager partitionManager = partitionManager(catalog); Throwable thrown = @@ -318,6 +242,7 @@ void testMissingTableStillFailsAfterFallback() throws Exception { .hasMessageContaining("list partitions") .hasMessageContaining("catalog_partition_db.catalog_partition_table"); assertThat(thrown.getCause()).isSameAs(notExist); + verify(catalog, never()).listPartitionsPaged(any(), any(), any(), any()); } // ------------------------------------------------------------------------ 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 aa4c9c7a7ea3..de2127f4939a 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 @@ -22,6 +22,7 @@ import org.apache.paimon.catalog.Catalog; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; @@ -40,6 +41,7 @@ import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; import javax.annotation.Nullable; @@ -74,11 +76,6 @@ class CatalogManagedPartitionScanTest { @Test void testLeadingPatternResidualFilterAndUnregisteredDirectory() throws Exception { Catalog catalog = mock(Catalog.class); - when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%"))) - .thenReturn( - new PagedList<>( - Arrays.asList(partition("2025", "10"), partition("2025", "11")), - null)); // The residual predicate beyond the prefix goes to the filter endpoint; the filter is a // hint, so returning a superset here is legal and the plan still filters per partition. when(catalog.listPartitionsByFilterPaged( @@ -109,13 +106,47 @@ void testLeadingPatternResidualFilterAndUnregisteredDirectory() throws Exception assertThat(plannedFiles).containsExactly(novemberFile); assertThat(plannedFiles).doesNotContain(octoberFile, unregisteredFile); assertThat(fileIO.listedPaths).containsExactly(new Path(tablePath, "year=2025/month=11")); + ArgumentCaptor pushedFilter = ArgumentCaptor.forClass(Predicate.class); verify(catalog) .listPartitionsByFilterPaged( eq(IDENTIFIER), - any(Predicate.class), + pushedFilter.capture(), eq(1000), isNull(), eq("year=2025/%")); + assertThat(pushedFilter.getValue().test(GenericRow.of(2025, 11))).isTrue(); + assertThat(pushedFilter.getValue().test(GenericRow.of(2025, 10))).isFalse(); + } + + @Test + void testPushesFilterWithoutLeadingPrefixAndAppliesResidualFilter() throws Exception { + TrackingLocalFileIO fileIO = new TrackingLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + Path octoberFile = writeDataFile(fileIO, tablePath, "year=2025/month=10"); + Path novemberFile = writeDataFile(fileIO, tablePath, "year=2025/month=11"); + FormatTable table = + createTable( + fileIO, + tablePath, + recordingCatalog( + Arrays.asList(partition("2025", "10"), partition("2025", "11"))), + false); + Predicate predicate = new PredicateBuilder(table.partitionType()).greaterThan(1, 10); + PartitionPredicate filter = + PartitionPredicate.fromPredicate(table.partitionType(), predicate); + + List plannedFiles = + plannedFiles(new FormatTableScan(table, filter, null).plan().splits()); + + assertThat(plannedFiles).containsExactly(novemberFile); + assertThat(plannedFiles).doesNotContain(octoberFile); + assertThat(fileIO.listedPaths).containsExactly(new Path(tablePath, "year=2025/month=11")); + assertThat(requestedPrefixes).containsExactly(Collections.emptyMap()); + assertThat(requestedFilters).hasSize(1); + Predicate pushedFilter = requestedFilters.get(0); + assertThat(pushedFilter).isNotNull(); + assertThat(pushedFilter.test(GenericRow.of(2025, 11))).isTrue(); + assertThat(pushedFilter.test(GenericRow.of(2025, 10))).isFalse(); } @Test @@ -325,6 +356,7 @@ void testDotDotValueInKeyValueLayoutRemainsVisible() throws Exception { } private final List> requestedPrefixes = new ArrayList<>(); + private final List requestedFilters = new ArrayList<>(); private FormatTablePartitionManager partitionManager(Catalog catalog) { return FormatTablePartitionManager.create( @@ -334,11 +366,13 @@ private FormatTablePartitionManager partitionManager(Catalog catalog) { /** Records the prefix the scan pushes down and answers from a fixed partition list. */ private FormatTablePartitionManager recordingCatalog(List partitions) { List> prefixes = requestedPrefixes; + List filters = requestedFilters; return new FormatTablePartitionManager() { @Override public List listPartitions( Map prefix, @Nullable Predicate filter) { prefixes.add(new LinkedHashMap<>(prefix)); + filters.add(filter); List matching = new ArrayList<>(); for (Partition partition : partitions) { boolean matches = true; diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala index 0c0035373e03..b8b8d53e21b0 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala @@ -25,14 +25,12 @@ import org.apache.paimon.rest.responses.ConfigResponse import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap import org.apache.spark.SparkConf -import org.apache.spark.sql.Row -import org.assertj.core.api.Assertions import java.util.UUID class PaimonSparkTestWithRestCatalogBase extends PaimonSparkTestBase { - protected var restCatalogServer: RESTCatalogServer = _ + private var restCatalogServer: RESTCatalogServer = _ private var serverUrl: String = _ protected var warehouse: String = _ private val initToken = "init_token" @@ -62,6 +60,12 @@ class PaimonSparkTestWithRestCatalogBase extends PaimonSparkTestBase { } } + protected def clearFilteredListingRequests(): Unit = + restCatalogServer.clearReceivedListPartitionsByFilterRequests() + + protected def filteredListingWasRequested: Boolean = + restCatalogServer.hasReceivedListPartitionsByFilterRequest + override protected def sparkConf: SparkConf = { super.sparkConf .set("spark.sql.catalog.paimon.metastore", RESTCatalogFactory.IDENTIFIER) diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionLoadTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionTest.scala similarity index 62% rename from paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionLoadTest.scala rename to paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionTest.scala index 9e39d195659e..191fe2cb46b4 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionLoadTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionTest.scala @@ -20,9 +20,10 @@ package org.apache.paimon.spark.format import org.apache.paimon.spark.{PaimonSparkTestWithRestCatalogBase, SparkCatalog} +import org.apache.spark.sql.Row import org.apache.spark.sql.connector.catalog.{Identifier => SparkIdentifier} -class CatalogManagedPartitionLoadTest extends PaimonSparkTestWithRestCatalogBase { +class CatalogManagedPartitionTest extends PaimonSparkTestWithRestCatalogBase { test( "a Format Table with catalog-managed partitions loaded through SparkCatalog carries its partition manager") { @@ -43,4 +44,27 @@ class CatalogManagedPartitionLoadTest extends PaimonSparkTestWithRestCatalogBase assert(sparkTable.partitionManager != null) } } + + test("catalog-managed Format Table sends partition predicate to REST filtered listing") { + val tableName = "catalog_partition_filter_pushdown" + withTable(tableName) { + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, year INT, month INT) + |USING CSV + |PARTITIONED BY (year, month) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + sql(s"""INSERT INTO $tableName VALUES + |(1, 'a', 2024, 12), (2, 'b', 2025, 9), (3, 'c', 2025, 11), (4, 'd', 2025, 12) + |""".stripMargin) + + clearFilteredListingRequests() + checkAnswer( + sql(s"SELECT id FROM $tableName WHERE year = 2025 AND month > 10 ORDER BY id"), + Seq(Row(3), Row(4))) + + assert(filteredListingWasRequested, "the filtered listing endpoint was not called") + } + } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ManagedFormatTablePartitionPruningTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ManagedFormatTablePartitionPruningTest.scala deleted file mode 100644 index 959336c6f0fe..000000000000 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ManagedFormatTablePartitionPruningTest.scala +++ /dev/null @@ -1,80 +0,0 @@ -/* - * 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.spark.sql - -import org.apache.paimon.predicate.{LeafPredicate, Predicate, PredicateBuilder} -import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase -import org.apache.paimon.utils.JsonSerdeUtil - -import org.apache.spark.sql.Row - -import scala.collection.JavaConverters._ - -/** - * End-to-end guard of the pushdown chain: a Spark SQL partition predicate on a catalog-managed - * Format Table must reach the REST `partitions/list-by-filter` endpoint, not only produce correct - * rows (the client-side residual filter would mask a broken transport path). - */ -class ManagedFormatTablePartitionPruningTest extends PaimonSparkTestWithRestCatalogBase { - - test("Format Table: partition predicate reaches the filtered listing endpoint") { - val tableName = "pruning_endpoint_t" - withTable(tableName) { - sql(s"""CREATE TABLE $tableName (id INT, payload STRING, year INT, month INT) - |USING CSV - |PARTITIONED BY (year, month) - |TBLPROPERTIES ( - | 'format-table.implementation' = 'paimon', - | 'metastore.partitioned-table' = 'true') - |""".stripMargin) - sql(s"""INSERT INTO $tableName VALUES - |(1, 'a', 2024, 12), (2, 'b', 2025, 9), (3, 'c', 2025, 11), (4, 'd', 2025, 12) - |""".stripMargin) - - restCatalogServer.receivedListPartitionsByFilterRequests.clear() - checkAnswer( - sql(s"SELECT id FROM $tableName WHERE year = 2025 AND month > 10 ORDER BY id"), - Seq(Row(3), Row(4))) - - val requests = restCatalogServer.receivedListPartitionsByFilterRequests - assert(!requests.isEmpty, "the filtered listing endpoint was not called") - val request = requests.get(requests.size() - 1) - assert(request.getFilter != null && request.getFilter.nonEmpty) - assert( - request.getPartitionNamePattern != null - && request.getPartitionNamePattern.startsWith("year=2025"), - s"unexpected pattern ${request.getPartitionNamePattern}" - ) - - // The filter restores to the pushed predicate (the wire form round-trips) and uses the - // flat encoding, no pretty-print newlines. - val restored = JsonSerdeUtil.fromJson(request.getFilter, classOf[Predicate]) - val leafFields = PredicateBuilder - .splitAnd(restored) - .asScala - .collect { case leaf: LeafPredicate => leaf.fieldNames().asScala } - .flatten - .toSet - assert( - leafFields.contains("year") && leafFields.contains("month"), - s"restored filter misses pushed fields: $restored") - assert(!request.getFilter.contains("\n"), "filter is not flat-encoded") - } - } -} From 5eeed6f13e0cfab1ef2b2abac78646e149616eff Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 22 Jul 2026 11:08:28 +0800 Subject: [PATCH 4/4] [docs][rest] Document partitions list-by-filter API --- docs/static/rest-catalog-open-api.yaml | 62 ++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/docs/static/rest-catalog-open-api.yaml b/docs/static/rest-catalog-open-api.yaml index 3956a9280649..51def16c4357 100644 --- a/docs/static/rest-catalog-open-api.yaml +++ b/docs/static/rest-catalog-open-api.yaml @@ -1145,6 +1145,50 @@ paths: $ref: '#/components/responses/TableNotExistErrorResponse' "500": $ref: '#/components/responses/ServerErrorResponse' + /v1/{prefix}/databases/{database}/tables/{table}/partitions/list-by-filter: + post: + tags: + - partition + summary: List partitions by filter + description: Lists a page of partitions using a serialized Paimon partition predicate. The response may be a superset of the matching partitions, so clients must re-evaluate the predicate and continue pagination while nextPageToken is non-empty, even if the current page contains no partitions. + operationId: listPartitionsByFilter + parameters: + - name: prefix + in: path + required: true + schema: + type: string + - name: database + in: path + required: true + schema: + type: string + - name: table + in: path + required: true + schema: + type: string + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/ListPartitionsByFilterRequest' + responses: + "200": + description: OK + content: + application/json: + schema: + $ref: '#/components/schemas/ListPartitionsResponse' + "400": + $ref: '#/components/responses/BadRequestErrorResponse' + "401": + $ref: '#/components/responses/UnauthorizedErrorResponse' + "404": + $ref: '#/components/responses/TableNotExistErrorResponse' + "500": + $ref: '#/components/responses/ServerErrorResponse' /v1/{prefix}/databases/{database}/tables/{table}/branches: get: tags: @@ -2522,6 +2566,24 @@ components: type: object additionalProperties: type: string + ListPartitionsByFilterRequest: + type: object + required: + - filter + properties: + filter: + type: string + description: JSON serialization of a Paimon Predicate tree over the partition columns. + partitionNamePattern: + type: string + description: A SQL LIKE pattern (%) for partition names. Currently, only prefix matching is supported. + maxResults: + type: integer + format: int32 + description: Maximum number of matching partitions to return in this page. A missing or zero value uses the server default. + pageToken: + type: string + description: Token returned by the previous page. Omit for the first page. CreateDatabaseResponse: type: object properties: