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: 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..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 @@ -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(), response.getNextPageToken()); + } + 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..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 @@ -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,32 @@ 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 + */ + default PagedList listPartitionsByFilterPaged( + Identifier identifier, + Predicate predicate, + @Nullable Integer maxResults, + @Nullable String pageToken, + @Nullable String partitionNamePattern) + throws TableNotExistException { + return listPartitionsPaged(identifier, maxResults, pageToken, partitionNamePattern); + } + /** * 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/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java index 7085135f17ff..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 @@ -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,68 @@ 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())); - } - } 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); + if (filter != null) { + return collectAllPages( + prefix, + pageToken -> + catalog.listPartitionsByFilterPaged( + identifier, + filter, + REQUEST_SIZE, + pageToken, + pattern)); + } + 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) { + for (Partition partition : page.getElements()) { + if (matchesPrefix(partition, prefix)) { + partitions.add(partition); + } + } + } + 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)); + return Collections.unmodifiableList(partitions); + } + + /** A single page of a partition listing. */ + private interface PageSupplier { + PagedList page(@Nullable String pageToken) throws Exception; + } + @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..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; @@ -258,19 +260,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,11 +286,45 @@ 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); } + @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 3a0e1539eea1..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 @@ -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,12 +153,15 @@ 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; 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; @@ -193,6 +202,13 @@ public class RESTCatalogServer { private final Map databaseStore = new HashMap<>(); private final Map tableMetadataStore = new HashMap<>(); + + 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<>(); @@ -279,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); } @@ -459,6 +494,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 +551,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 +562,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 +1989,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); + 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); + } + 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 (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<>(); + 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..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 @@ -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; @@ -372,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)); @@ -1877,6 +1889,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")); + 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(); + 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()) { + for (Map partitionSpec : partitionSpecs) { + write.write(GenericRow.of(0, BinaryString.fromString(partitionSpec.get("dt")))); + } + 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. + PagedList firstPage = + catalog.listPartitionsByFilterPaged(identifier, range, 2, null, "dt=2025%"); + assertThat(firstPage.getElements()) + .extracting(partition -> partition.spec().get("dt")) + .containsExactly("20250103", "20250102"); + assertThat(firstPage.getNextPageToken()).isEqualTo("dt=20250102"); + + 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 + // 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()) + .extracting(Partition::spec) + .containsExactlyInAnyOrderElementsOf(partitionSpecs); + } + @Test public void testListPartitionsPagedWithMultiLevel() throws Exception { if (!supportPartitions()) { 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..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 @@ -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; @@ -63,9 +70,16 @@ 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. */ + 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 +98,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 +115,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()); @@ -115,16 +129,122 @@ 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); - 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 + // ------------------------------------------------------------------------ + + @Test + 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<>(Arrays.asList(partition("2024", "12"), 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 testCatalogDefaultFilteredListingDelegatesToPlainListing() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition only = partition("2025", "01"); + when(catalog.listPartitionsByFilterPaged(any(), any(), any(), any(), any())) + .thenCallRealMethod(); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Collections.singletonList(only), null)); + + PagedList page = + catalog.listPartitionsByFilterPaged( + IDENTIFIER, FILTER, REQUEST_SIZE, "page-2", "year=2025/%"); + + assertThat(page.getElements()).containsExactly(only); + verify(catalog) + .listPartitionsPaged( + eq(IDENTIFIER), eq(REQUEST_SIZE), eq("page-2"), eq("year=2025/%")); + } + + @Test + void testOtherFilteredFailuresDoNotFallBack() throws Exception { + Catalog catalog = mock(Catalog.class); + Catalog.TableNoPermissionException failure = + new Catalog.TableNoPermissionException(IDENTIFIER); + 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 testMissingTableDoesNotTriggerPlainListing() throws Exception { + Catalog catalog = mock(Catalog.class); + Catalog.TableNotExistException notExist = new Catalog.TableNotExistException(IDENTIFIER); + when(catalog.listPartitionsByFilterPaged(any(), 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); + verify(catalog, never()).listPartitionsPaged(any(), any(), any(), any()); + } + // ------------------------------------------------------------------------ // prefix pushdown and client side filtering // ------------------------------------------------------------------------ @@ -135,7 +255,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 +267,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 +281,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 +294,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 +389,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 +414,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 +426,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 +444,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 +481,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..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,9 @@ 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; import java.io.FileNotFoundException; import java.io.IOException; @@ -54,9 +58,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. */ @@ -70,7 +76,14 @@ class CatalogManagedPartitionScanTest { @Test void testLeadingPatternResidualFilterAndUnregisteredDirectory() throws Exception { Catalog catalog = mock(Catalog.class); - when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%"))) + // 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")), @@ -93,6 +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), + 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 @@ -302,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( @@ -311,10 +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) { + 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-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..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,8 +25,6 @@ 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 @@ -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/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/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/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) } }