[rest] Add tagNamePrefix definition for listTagsPaged

docs/static/rest-catalog-open-api.yaml

@@ -1147,6 +1147,11 @@ paths:
           in: query
           schema:
             type: string
+        - name: tagNamePrefix
+          description: A prefix for tag names. All tags will be returned if not set or empty.
+          in: query
+          schema:
+            type: string
       responses:
         "200":
           description: OK

paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java

@@ -147,6 +147,7 @@ public class RESTApi {
     public static final String VIEW_NAME_PATTERN = "viewNamePattern";
     public static final String FUNCTION_NAME_PATTERN = "functionNamePattern";
     public static final String PARTITION_NAME_PATTERN = "partitionNamePattern";
+    public static final String TAG_NAME_PREFIX = "tagNamePrefix";
 
     public static final long TOKEN_EXPIRATION_SAFE_TIME_MILLIS = 3_600_000L;
 
@@ -917,18 +918,23 @@ public void createTag(
      *     max results.
      * @param pageToken Optional parameter indicating the next page token allows list to be start
      *     from a specific point.
+     * @param tagNamePrefix A prefix for tag names. All tags will be returned if not set or empty.
      * @return {@link PagedList}: elements and nextPageToken.
      * @throws NoSuchResourceException Exception thrown on HTTP 404 means the table not exists
      * @throws ForbiddenException Exception thrown on HTTP 403 means don't have the permission for
      *     this table
      */
     public PagedList<String> listTagsPaged(
-            Identifier identifier, @Nullable Integer maxResults, @Nullable String pageToken) {
+            Identifier identifier,
+            @Nullable Integer maxResults,
+            @Nullable String pageToken,
+            @Nullable String tagNamePrefix) {
         ListTagsResponse response =
                 client.get(
                         resourcePaths.tags(
                                 identifier.getDatabaseName(), identifier.getObjectName()),
-                        buildPagedQueryParams(maxResults, pageToken),
+                        buildPagedQueryParams(
+                                maxResults, pageToken, Pair.of(TAG_NAME_PREFIX, tagNamePrefix)),
                         ListTagsResponse.class,
                         restAuthFunction);
         List<String> tags = response.tags();

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -521,7 +521,10 @@ public void createTag(
 
     @Override
     public PagedList<String> listTagsPaged(
-            Identifier identifier, @Nullable Integer maxResults, @Nullable String pageToken)
+            Identifier identifier,
+            @Nullable Integer maxResults,
+            @Nullable String pageToken,
+            @Nullable String tagNamePrefix)
             throws TableNotExistException {
         throw new UnsupportedOperationException();
     }

paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java

@@ -631,7 +631,7 @@ default boolean supportsListTableByType() {
      *   <li>{@link #listBranches(Identifier)}.
      *   <li>{@link #getTag(Identifier, String)}.
      *   <li>{@link #createTag(Identifier, String, Long, String, boolean)}.
-     *   <li>{@link #listTagsPaged(Identifier, Integer, String)}.
+     *   <li>{@link #listTagsPaged(Identifier, Integer, String, String)}.
      *   <li>{@link #deleteTag(Identifier, String)}.
      * </ul>
      */
@@ -830,6 +830,7 @@ void createTag(
      *     max results.
      * @param pageToken Optional parameter indicating the next page token allows list to be start
      *     from a specific point.
+     * @param tagNamePrefix A prefix for tag names. All tags will be returned if not set or empty.
      * @return a list of the names of tags with provided page size in this table and next page
      *     token, or a list of the names of all tags in this table if the catalog does not {@link
      *     #supportsListObjectsPaged()}.
@@ -838,7 +839,10 @@ void createTag(
      *     #supportsVersionManagement()} or it does not {@link #supportsListByPattern()}
      */
     PagedList<String> listTagsPaged(
-            Identifier identifier, @Nullable Integer maxResults, @Nullable String pageToken)
+            Identifier identifier,
+            @Nullable Integer maxResults,
+            @Nullable String pageToken,
+            @Nullable String tagNamePrefix)
             throws TableNotExistException;
 
     /**

paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java

@@ -252,9 +252,12 @@ public void createTag(
 
     @Override
     public PagedList<String> listTagsPaged(
-            Identifier identifier, @Nullable Integer maxResults, @Nullable String pageToken)
+            Identifier identifier,
+            @Nullable Integer maxResults,
+            @Nullable String pageToken,
+            @Nullable String tagNamePrefix)
             throws TableNotExistException {
-        return wrapped.listTagsPaged(identifier, maxResults, pageToken);
+        return wrapped.listTagsPaged(identifier, maxResults, pageToken, tagNamePrefix);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -1053,10 +1053,13 @@ public void createTag(
      */
     @Override
     public PagedList<String> listTagsPaged(
-            Identifier identifier, @Nullable Integer maxResults, @Nullable String pageToken)
+            Identifier identifier,
+            @Nullable Integer maxResults,
+            @Nullable String pageToken,
+            @Nullable String tagNamePrefix)
             throws TableNotExistException {
         try {
-            return api.listTagsPaged(identifier, maxResults, pageToken);
+            return api.listTagsPaged(identifier, maxResults, pageToken, tagNamePrefix);
         } catch (NoSuchResourceException e) {
             throw new TableNotExistException(identifier);
         } catch (ForbiddenException e) {

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java

@@ -153,6 +153,7 @@
 import static org.apache.paimon.rest.RESTApi.PARTITION_NAME_PATTERN;
 import static org.apache.paimon.rest.RESTApi.TABLE_NAME_PATTERN;
 import static org.apache.paimon.rest.RESTApi.TABLE_TYPE;
+import static org.apache.paimon.rest.RESTApi.TAG_NAME_PREFIX;
 import static org.apache.paimon.rest.RESTApi.VIEW_NAME_PATTERN;
 import static org.apache.paimon.rest.ResourcePaths.FUNCTIONS;
 import static org.apache.paimon.rest.ResourcePaths.FUNCTION_DETAILS;
@@ -1736,6 +1737,16 @@ private MockResponse tagApiHandle(
                         // GET /v1/{prefix}/databases/{database}/tables/{table}/tags
                         // Page list tags
                         List<String> tags = new ArrayList<>(tagManager.allTagNames());
+
+                        // Filter by tagNamePrefix if provided
+                        String tagNamePrefix = parameters.get(TAG_NAME_PREFIX);
+                        if (StringUtils.isNotEmpty(tagNamePrefix)) {
+                            tags =
+                                    tags.stream()
+                                            .filter(tag -> matchNamePattern(tag, tagNamePrefix))
+                                            .collect(Collectors.toList());
+                        }
+
                         if (tags.isEmpty()) {
                             response = new ListTagsResponse(Collections.emptyList(), null);
                             return mockResponse(response, 200);

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java

@@ -1966,15 +1966,20 @@ void testTags() throws Exception {
                 SnapshotNotExistException.class,
                 () -> restCatalog.createTag(identifier, "my_tag_v3", 99999L, null, false));
 
-        // Test listTags
-        PagedList<String> tags = restCatalog.listTagsPaged(identifier, null, "my_tag");
+        // Test listTags for pageToken
+        PagedList<String> tags = restCatalog.listTagsPaged(identifier, 1, null, null);
+        tags = restCatalog.listTagsPaged(identifier, 1, tags.getNextPageToken(), null);
         assertThat(tags.getElements()).containsExactlyInAnyOrder("my_tag_v2");
-        tags = restCatalog.listTagsPaged(identifier, null, null);
+        tags = restCatalog.listTagsPaged(identifier, null, null, null);
         assertThat(tags.getElements()).containsExactlyInAnyOrder("my_tag", "my_tag_v2");
 
+        // Test listTags for tagNamePrefix
+        tags = restCatalog.listTagsPaged(identifier, 1, null, "my_tag_v2");
+        assertThat(tags.getElements()).containsExactlyInAnyOrder("my_tag_v2");
+
         // Test deleteTag
         restCatalog.deleteTag(identifier, "my_tag");
-        tags = restCatalog.listTagsPaged(identifier, null, null);
+        tags = restCatalog.listTagsPaged(identifier, null, null, null);
         assertThat(tags.getElements()).containsExactlyInAnyOrder("my_tag_v2");
 
         // Test deleteTag with non-existent tag
@@ -1988,7 +1993,7 @@ void testTags() throws Exception {
 
         // Delete remaining tag
         restCatalog.deleteTag(identifier, "my_tag_v2");
-        tags = restCatalog.listTagsPaged(identifier, null, null);
+        tags = restCatalog.listTagsPaged(identifier, null, null, null);
         assertThat(tags.getElements()).isEmpty();
     }