This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b52f55f780 [core][rest] Push Format Table partition filters to REST
catalog (#8785)
b52f55f780 is described below
commit b52f55f7801e16cdfd64b2fc935b3d3f14257af8
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Wed Jul 22 11:38:02 2026 +0800
[core][rest] Push Format Table partition filters to REST catalog (#8785)
---
docs/static/rest-catalog-open-api.yaml | 62 +++++++++
.../main/java/org/apache/paimon/rest/RESTApi.java | 43 ++++++
.../java/org/apache/paimon/rest/ResourcePaths.java | 12 ++
.../requests/ListPartitionsByFilterRequest.java | 91 +++++++++++++
.../java/org/apache/paimon/catalog/Catalog.java | 27 ++++
.../org/apache/paimon/catalog/DelegateCatalog.java | 13 ++
.../java/org/apache/paimon/rest/RESTCatalog.java | 28 ++++
.../format/CatalogFormatTablePartitionManager.java | 84 +++++++++---
.../table/format/FormatTablePartitionManager.java | 10 +-
.../paimon/table/format/FormatTableScan.java | 20 +--
.../apache/paimon/rest/MockRESTCatalogTest.java | 52 ++++++-
.../org/apache/paimon/rest/RESTApiJsonTest.java | 28 ++++
.../org/apache/paimon/rest/RESTCatalogServer.java | 123 ++++++++++++++++-
.../org/apache/paimon/rest/RESTCatalogTest.java | 102 ++++++++++++++
.../CatalogFormatTablePartitionManagerTest.java | 149 +++++++++++++++++++--
.../format/CatalogManagedPartitionScanTest.java | 62 ++++++++-
.../spark/format/FormatTablePartitionRepair.java | 3 +-
.../paimon/spark/format/PaimonFormatTable.scala | 2 +-
.../format/FormatTablePartitionRepairTest.java | 9 +-
.../spark/PaimonSparkTestWithRestCatalogBase.scala | 8 +-
.../FormatTablePartitionDdlPlanningTest.scala | 38 ++++--
...est.scala => CatalogManagedPartitionTest.scala} | 26 +++-
.../FormatTablePartitionManagementTest.scala | 45 +++++--
.../CatalogManagedPartitionMsckRepairTest.scala | 7 +-
24 files changed, 959 insertions(+), 85 deletions(-)
diff --git a/docs/static/rest-catalog-open-api.yaml
b/docs/static/rest-catalog-open-api.yaml
index 3956a92806..51def16c43 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 4d321ed279..e8e1ff96c8 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.CreateTagRequest;
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 class RESTApi {
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.
*
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 0ab36e348c..ae09705467 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 class ResourcePaths {
"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 0000000000..cd0fa058c4
--- /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 8e4ad8d518..fd477ffcf1 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.Function;
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 @@ public interface Catalog extends AutoCloseable {
@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.
*
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 290849024f..10ae54ea45 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.Function;
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 abstract class DelegateCatalog implements Catalog {
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)
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 efef63fc92..86256a9342 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.function.FunctionChange;
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.Table;
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 class RESTCatalog implements Catalog {
}
}
+ @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)
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 7085135f17..ed266db2a5 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.Catalog;
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 @@ class CatalogFormatTablePartitionManager implements
FormatTablePartitionManager
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()) {
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 64b643fba2..259f88bbfb 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.annotation.Experimental;
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.
+ *
+ * <p>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<Partition> listPartitions(Map<String, String> prefix);
+ List<Partition> listPartitions(Map<String, String> prefix, @Nullable
Predicate filter);
/** Return those of the given complete partition specs that are
registered. */
List<Partition> listPartitionsByNames(List<Map<String, String>>
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 44908749ca..2dee6fe123 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 class FormatTableScan implements InnerTableScan {
@Override
public List<PartitionEntry> listPartitionEntries() {
if (partitionManager != null) {
- List<Partition> partitions =
partitionManager.listPartitions(Collections.emptyMap());
+ List<Partition> partitions =
+ partitionManager.listPartitions(Collections.emptyMap(),
null);
if (partitions.isEmpty()) {
warnIfFilesystemPartitionsExist();
}
@@ -228,13 +229,17 @@ public class FormatTableScan implements InnerTableScan {
List<Pair<LinkedHashMap<String, String>, Path>> findPartitions() {
if (partitionManager != null) {
- Map<String, String> prefix = leadingEqualityPrefix();
- List<Partition> partitions =
partitionManager.listPartitions(prefix);
- if (partitions.isEmpty() && prefix.isEmpty()) {
+ Optional<Predicate> extracted =
extractPartitionPredicate(partitionFilter);
+ Map<String, String> 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<Partition> 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 @@ public class FormatTableScan implements InnerTableScan {
* to the partition catalog. Empty when there is no predicate or it does
not start with
* equalities on the leading partition keys.
*/
- private Map<String, String> leadingEqualityPrefix() {
- Optional<Predicate> predicate =
extractPartitionPredicate(partitionFilter);
+ private Map<String, String> leadingEqualityPrefix(Optional<Predicate>
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 6cc8f70420..ff3736d889 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.Path;
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 org.junit.jupiter.api.Test;
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 @@ class MockRESTCatalogTest extends RESTCatalogTest {
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<String, String> 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 @@ class MockRESTCatalogTest extends RESTCatalogTest {
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<Partition> firstPage =
+ restCatalog.listPartitionsByFilterPaged(identifier, predicate,
1, null, "dt=2026%");
+ assertThat(firstPage.getElements()).isEmpty();
+ assertThat(firstPage.getNextPageToken()).isEqualTo("p2");
+
+ PagedList<Partition> 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 @@ class MockRESTCatalogTest extends RESTCatalogTest {
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 bdf2b5f82c..7508e00473 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.CreatePartitionsRequest;
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 class RESTApiJsonTest {
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<String, Object> 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<String, String> 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 3a0e1539ee..6796e1ddf7 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.Identifier;
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.CreateTableRequest;
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.Instant;
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.Comparator;
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<String, Database> databaseStore = new HashMap<>();
private final Map<String, TableMetadata> tableMetadataStore = new
HashMap<>();
+
+ private final List<ListPartitionsByFilterRequest>
receivedListPartitionsByFilterRequests =
+ new java.util.concurrent.CopyOnWriteArrayList<>();
+
+ private final Queue<ListPartitionsResponse>
scriptedListPartitionsByFilterResponses =
+ new ConcurrentLinkedQueue<>();
+
private final Map<String, List<Partition>> tablePartitionsStore = new
HashMap<>();
private final Map<String, View> viewStore = new HashMap<>();
private final Map<String, TableSnapshot> tableLatestSnapshotStore = new
HashMap<>();
@@ -279,6 +295,25 @@ public class RESTCatalogServer {
this.partitionListingSupported = partitionListingSupported;
}
+ public void clearReceivedListPartitionsByFilterRequests() {
+ receivedListPartitionsByFilterRequests.clear();
+ }
+
+ public List<ListPartitionsByFilterRequest>
getReceivedListPartitionsByFilterRequests() {
+ return Collections.unmodifiableList(
+ new ArrayList<>(receivedListPartitionsByFilterRequests));
+ }
+
+ public boolean hasReceivedListPartitionsByFilterRequest() {
+ return !receivedListPartitionsByFilterRequests.isEmpty();
+ }
+
+ public void enqueueListPartitionsByFilterResponse(
+ @Nullable List<Partition> partitions, @Nullable String
nextPageToken) {
+ scriptedListPartitionsByFilterResponses.add(
+ new ListPartitionsResponse(partitions, nextPageToken));
+ }
+
public void addNoPermissionDatabase(String database) {
noPermissionDatabases.add(database);
}
@@ -459,6 +494,11 @@ public class RESTCatalogServer {
&&
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 @@ public class RESTCatalogServer {
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 @@ public class RESTCatalogServer {
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 @@ public class RESTCatalogServer {
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<Partition> 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<String, String> 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<String, String> spec,
+ @Nullable RowType partitionType,
+ String defaultPartName) {
+ if (predicate == null || partitionType == null) {
+ return true;
+ }
+ try {
+ LinkedHashMap<String, String> 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<String, String> 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 0e386e8b10..05376d3585 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.source.TableRead;
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 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
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 abstract class RESTCatalogTest extends
CatalogTestBase {
() -> catalog.listPartitionsPaged(identifier, null, null,
"dt=01%01"));
}
+ @Test
+ public void testListPartitionsByFilterPaged() throws Exception {
+ if (!supportPartitions()) {
+ return;
+ }
+
+ String databaseName = "partitions_filter_db";
+ List<Map<String, String>> 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<String, String> 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<Partition> 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<Partition> 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 d7a5d97e69..7d246bf4f5 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.PagedList;
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.eq;
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<String> 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 @@ class CatalogFormatTablePartitionManagerTest {
new PagedList<>(Collections.singletonList(second),
null));
List<Partition> partitions =
-
partitionManager(catalog).listPartitions(Collections.emptyMap());
+
partitionManager(catalog).listPartitions(Collections.emptyMap(), null);
assertThat(partitions).containsExactly(first, second);
ArgumentCaptor<String> pageTokens =
ArgumentCaptor.forClass(String.class);
@@ -101,7 +115,7 @@ class CatalogFormatTablePartitionManagerTest {
.thenReturn(new PagedList<>(Collections.singletonList(only),
""));
List<Partition> 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 @@ class CatalogFormatTablePartitionManagerTest {
.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<Partition> partitions =
+ partitionManager(catalog)
+ .listPartitions(Collections.singletonMap("year",
"2025"), FILTER);
+
+ assertThat(partitions).containsExactly(first, second);
+ ArgumentCaptor<String> 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<Partition> 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<Partition> 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 @@ class CatalogFormatTablePartitionManagerTest {
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 @@ class CatalogFormatTablePartitionManagerTest {
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 @@ class CatalogFormatTablePartitionManagerTest {
.thenReturn(new PagedList<>(Arrays.asList(matching, other),
null));
List<Partition> partitions =
-
partitionManager(catalog).listPartitions(Collections.singletonMap("year",
"2025"));
+ partitionManager(catalog)
+ .listPartitions(Collections.singletonMap("year",
"2025"), null);
assertThat(partitions).containsExactly(matching);
}
@@ -173,7 +294,7 @@ class CatalogFormatTablePartitionManagerTest {
// "month" without "year" cannot be expressed as a partition path
prefix.
Map<String, String> prefix = Collections.singletonMap("month", "10");
- assertThatThrownBy(() -> partitionManager.listPartitions(prefix))
+ assertThatThrownBy(() -> partitionManager.listPartitions(prefix, null))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("leading prefix");
}
@@ -268,7 +389,7 @@ class CatalogFormatTablePartitionManagerTest {
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 @@ class CatalogFormatTablePartitionManagerTest {
.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 @@ class CatalogFormatTablePartitionManagerTest {
FormatTablePartitionManager partitionManager =
partitionManager(catalog);
Map<String, String> prefix = Collections.singletonMap("month", "10");
- assertThatThrownBy(() -> partitionManager.listPartitions(prefix))
+ assertThatThrownBy(() -> partitionManager.listPartitions(prefix, null))
.isInstanceOf(IllegalArgumentException.class);
verifyNoInteractions(catalog);
@@ -323,7 +444,7 @@ class CatalogFormatTablePartitionManagerTest {
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 @@ class CatalogFormatTablePartitionManagerTest {
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 da74ea3273..de2127f493 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.PagedList;
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.apache.paimon.utils.PartitionPathUtils;
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 java.util.stream.Collectors;
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 @@ class CatalogManagedPartitionScanTest {
assertThat(plannedFiles).containsExactly(novemberFile);
assertThat(plannedFiles).doesNotContain(octoberFile, unregisteredFile);
assertThat(fileIO.listedPaths).containsExactly(new Path(tablePath,
"year=2025/month=11"));
+ ArgumentCaptor<Predicate> 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<Path> 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 @@ class CatalogManagedPartitionScanTest {
}
private final List<Map<String, String>> requestedPrefixes = new
ArrayList<>();
+ private final List<Predicate> requestedFilters = new ArrayList<>();
private FormatTablePartitionManager partitionManager(Catalog catalog) {
return FormatTablePartitionManager.create(
@@ -311,10 +366,13 @@ class CatalogManagedPartitionScanTest {
/** Records the prefix the scan pushes down and answers from a fixed
partition list. */
private FormatTablePartitionManager recordingCatalog(List<Partition>
partitions) {
List<Map<String, String>> prefixes = requestedPrefixes;
+ List<Predicate> filters = requestedFilters;
return new FormatTablePartitionManager() {
@Override
- public List<Partition> listPartitions(Map<String, String> prefix) {
+ public List<Partition> listPartitions(
+ Map<String, String> prefix, @Nullable Predicate filter) {
prefixes.add(new LinkedHashMap<>(prefix));
+ filters.add(filter);
List<Partition> 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 4aa99b812c..d77968f248 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 @@ public class FormatTablePartitionRepair {
boolean addPartitions,
boolean dropPartitions) {
Set<Map<String, String>> registeredPartitions = new HashSet<>();
- for (Partition partition :
partitionManager.listPartitions(Collections.emptyMap())) {
+ for (Partition partition :
+ partitionManager.listPartitions(Collections.<String,
String>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 eecaed851b..6b5e5210f9 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
@@ -166,7 +166,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 5bf21ded18..bc6e24d219 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.FileStatus;
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.apache.paimon.types.RowType;
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 @@ class FormatTablePartitionRepairTest {
RecordingPartitionManager catalog =
new RecordingPartitionManager() {
@Override
- public List<Partition> listPartitions(Map<String, String>
prefix) {
+ public List<Partition> listPartitions(
+ Map<String, String> prefix, @Nullable Predicate
filter) {
listCount.incrementAndGet();
throw listFailure;
}
@@ -456,7 +460,8 @@ class FormatTablePartitionRepairTest {
}
@Override
- public List<Partition> listPartitions(Map<String, String> prefix) {
+ public List<Partition> listPartitions(
+ Map<String, String> prefix, @Nullable Predicate filter) {
requestedPrefixes.add(prefix);
List<Partition> partitions = new ArrayList<>(registered.size());
for (Map<String, String> 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 c880ba7f36..b8b8d53e21 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 7ef72caad0..7968c8d49c 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 9e39d19565..191fe2cb46 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 3cb44485de..7c36fd09be 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 3997341b45..24e42b9c86 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)
}
}
