[core][rest] Push Format Table partition filters to REST catalog

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<Partition> listPartitionsByNames(
         return partitions;
     }
 
+    /**
+     * List a page of partitions using a serialized partition predicate.
+     *
+     * <p>{@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<Partition> 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<Partition> partitions = response.getPartitions();
+        if (partitions == null) {
+            return new PagedList<>(emptyList(), response.getNextPageToken());
+        }
+        return new PagedList<>(partitions, response.getNextPageToken());
+    }
+
     /**
      * Create branch for table.
      *

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,

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;
+    }
+}

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<Partition> listPartitionsPaged(
             @Nullable String partitionNamePattern)
             throws TableNotExistException;
 
+    /**
+     * Get a page of partitions using {@code predicate} as a best-effort filter.
+     *
+     * <p>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<Partition> 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.
      *

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<Partition> listPartitionsPaged(
         return wrapped.listPartitionsPaged(identifier, maxResults, pageToken, partitionNamePattern);
     }
 
+    @Override
+    public PagedList<Partition> listPartitionsByFilterPaged(
+            Identifier identifier,
+            Predicate predicate,
+            Integer maxResults,
+            String pageToken,
+            String partitionNamePattern)
+            throws TableNotExistException {
+        return wrapped.listPartitionsByFilterPaged(
+                identifier, predicate, maxResults, pageToken, partitionNamePattern);
+    }
+
     @Override
     public List<Partition> listPartitionsByNames(
             Identifier identifier, List<Map<String, String>> partitions)

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<Partition> listPartitionsPaged(
         }
     }
 
+    @Override
+    public PagedList<Partition> 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<Partition> listPartitionsByNames(
             Identifier identifier, List<Map<String, String>> partitions)

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<Partition> listPartitions(Map<String, String> prefix) {
+    public List<Partition> listPartitions(Map<String, String> prefix, @Nullable Predicate filter) {
         LinkedHashMap<String, String> ordered = orderedPrefix(partitionKeys, prefix);
         checkArgument(
                 ordered.size() == prefix.size(),
@@ -73,31 +76,68 @@ public List<Partition> listPartitions(Map<String, String> prefix) {
         String pattern = PartitionPathUtils.buildPartitionNamePrefixPattern(partitionKeys, ordered);
         return execute(
                 catalog -> {
-                    List<Partition> partitions = new ArrayList<>();
-                    Set<String> seenPageTokens = new HashSet<>();
-                    String pageToken = null;
-                    do {
-                        PagedList<Partition> 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<Partition> collectAllPages(Map<String, String> prefix, PageSupplier pageSupplier)
+            throws Exception {
+        List<Partition> partitions = new ArrayList<>();
+        Set<String> seenPageTokens = new HashSet<>();
+        String pageToken = null;
+        do {
+            PagedList<Partition> page = pageSupplier.page(pageToken);
+            if (page == null) {
+                // A missing page cannot be told apart from an empty listing; failing loudly
+                // beats silently reading fewer partitions.
+                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<Partition> page(@Nullable String pageToken) throws Exception;
+    }
+
     @Override
     public List<Partition> listPartitionsByNames(List<Map<String, String>> partitions) {
         if (partitions.isEmpty()) {