[jdbc] Implement real paging and pattern filtering for listViewsPaged.

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

@@ -498,7 +498,8 @@ default List<String> listViews(String databaseName) throws DatabaseNotExistExcep
      * @param pageToken Optional parameter indicating the next page token allows list to be start
      *     from a specific point.
      * @param viewNamePattern A sql LIKE pattern (%) for view names. All views will be returned if
-     *     not set or empty. Currently, only prefix matching is supported.
+     *     not set or empty. Whether full LIKE semantics or only prefix matching is supported is
+     *     catalog-specific; the default implementation ignores the pattern.
      * @return a list of the names of views with provided page size in this database and next page
      *     token, or a list of the names of all views in this database if the catalog does not
      *     {@link #supportsListObjectsPaged()}.
@@ -525,7 +526,8 @@ default PagedList<String> listViewsPaged(
      * @param pageToken Optional parameter indicating the next page token allows list to be start
      *     from a specific point.
      * @param viewNamePattern A sql LIKE pattern (%) for view names. All view details will be
-     *     returned if not set or empty. Currently, only prefix matching is supported.
+     *     returned if not set or empty. Whether full LIKE semantics or only prefix matching is
+     *     supported is catalog-specific; the default implementation ignores the pattern.
      * @return a list of the view details with provided page size (@param maxResults) in this
      *     database and next page token, or a list of the details of all views in this database if
      *     the catalog does not {@link #supportsListObjectsPaged()}.
@@ -651,7 +653,10 @@ default void registerTable(Identifier identifier, String path)
 
     /**
      * Whether this catalog supports name pattern filter when list objects paged. If not,
-     * corresponding methods will throw exception if name pattern provided.
+     * corresponding methods will throw exception if name pattern provided. This flag is a
+     * catalog-wide default consulted by the base implementations; a specific list method may still
+     * honor a pattern by overriding the method directly (in which case it need not rely on this
+     * flag).
      *
      * <ul>
      *   <li>{@link #listDatabasesPaged(Integer, String, String)}.

paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java

@@ -65,6 +65,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.AbstractMap;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -1174,13 +1175,64 @@ public List<String> listViews(String databaseName) throws DatabaseNotExistExcept
                 databaseName);
     }
 
-    // TODO: Implement actual paging and pattern filtering
     @Override
     public PagedList<String> listViewsPaged(
             String databaseName, Integer maxResults, String pageToken, String viewNamePattern)
             throws DatabaseNotExistException {
-        CatalogUtils.validateNamePattern(this, viewNamePattern);
-        return new PagedList<>(listViews(databaseName), null);
+        if (CatalogUtils.isSystemDatabase(databaseName)) {
+            return new PagedList<>(Collections.emptyList(), null);
+        }
+
+        // Check if database exists
+        if (!JdbcUtils.databaseExists(connections, catalogKey, databaseName)) {
+            throw new DatabaseNotExistException(databaseName);
+        }
+
+        // CatalogUtils.validateNamePattern is intentionally NOT called here: this method supports
+        // pattern filtering directly via SQL LIKE. The catalog-wide supportsListByPattern flag
+        // remains false until other list methods (tables/databases) also gain pattern support, so
+        // that callers of those methods still get an explicit UnsupportedOperationException rather
+        // than silently unfiltered results. Unlike the default implementation, this override honors
+        // the pattern even though supportsListByPattern() is false.
+
+        // Per the Catalog contract, a null OR empty pattern means "no pattern": return all views.
+        boolean hasPattern = viewNamePattern != null && !viewNamePattern.isEmpty();
+        // pageToken is the last view name returned by the previous page (opaque to callers). The
+        // empty lower bound returns every view ordered by name.
+        String cursor = pageToken == null ? "" : pageToken;
+
+        String sql;
+        String[] args;
+        if (hasPattern) {
+            sql = JdbcUtils.LIST_VIEWS_PAGED_WITH_PATTERN_SQL;
+            args = new String[] {catalogKey, databaseName, viewNamePattern, cursor};
+        } else {
+            sql = JdbcUtils.LIST_VIEWS_PAGED_SQL;
+            args = new String[] {catalogKey, databaseName, cursor};
+        }
+
+        // Per the Catalog contract, maxResults == null OR 0 means "no paging": return all matching
+        // views, ordered, with no next page.
+        if (maxResults == null || maxResults == 0) {
+            List<String> views = fetch(row -> row.getString(JdbcUtils.VIEW_NAME), sql, args);
+            return new PagedList<>(views, null);
+        }
+
+        Preconditions.checkArgument(maxResults > 0, "maxResults must be positive when provided");
+        // Fetch one extra row to detect whether another page follows, without relying on a count.
+        List<String> views =
+                fetch(
+                        row -> row.getString(JdbcUtils.VIEW_NAME),
+                        sql + " LIMIT " + (maxResults + 1),
+                        args);
+        String nextPageToken = null;
+        if (views.size() > maxResults) {
+            // More pages remain. Drop the lookahead row and use the last returned name as the
+            // cursor for the next page.
+            views = new ArrayList<>(views.subList(0, maxResults));
+            nextPageToken = views.get(maxResults - 1);
+        }
+        return new PagedList<>(views, nextPageToken);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcUtils.java

@@ -411,6 +411,38 @@ JdbcViewConflictKind kind() {
                     + VIEW_DATABASE
                     + " = ?";
 
+    // Ordered view listing with an exclusive cursor (view_name > ?). A {@code LIMIT n} clause is
+    // appended by the caller when paging is requested.
+    static final String LIST_VIEWS_PAGED_SQL =
+            "SELECT "
+                    + VIEW_NAME
+                    + " FROM "
+                    + VIEW_TABLE_NAME
+                    + " WHERE "
+                    + CATALOG_KEY
+                    + " = ? AND "
+                    + VIEW_DATABASE
+                    + " = ? AND "
+                    + VIEW_NAME
+                    + " > ? ORDER BY "
+                    + VIEW_NAME;
+
+    static final String LIST_VIEWS_PAGED_WITH_PATTERN_SQL =
+            "SELECT "
+                    + VIEW_NAME
+                    + " FROM "
+                    + VIEW_TABLE_NAME
+                    + " WHERE "
+                    + CATALOG_KEY
+                    + " = ? AND "
+                    + VIEW_DATABASE
+                    + " = ? AND "
+                    + VIEW_NAME
+                    + " LIKE ? AND "
+                    + VIEW_NAME
+                    + " > ? ORDER BY "
+                    + VIEW_NAME;
+
     static final String INSERT_VIEW_SQL =
             "INSERT INTO "
                     + VIEW_TABLE_NAME

paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.jdbc;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.PagedList;
 import org.apache.paimon.TableType;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
@@ -50,6 +51,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.SQLIntegrityConstraintViolationException;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -60,6 +62,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.stream.Collectors;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -978,6 +981,170 @@ public void testListViewsFromSystemDatabase() throws Exception {
                 .isEmpty();
     }
 
+    @Override
+    @Test
+    public void testListViewsPaged() throws Exception {
+        String databaseName = "views_paged_db";
+        catalog.createDatabase(databaseName, false);
+
+        // Empty database returns an empty page with no next token.
+        PagedList<String> pagedViews = catalog.listViewsPaged(databaseName, null, null, null);
+        assertThat(pagedViews.getElements()).isEmpty();
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        View view = buildView(databaseName);
+        String[] viewNames = {"view1", "view2", "view3", "abd", "def", "opr"};
+        String[] sortedViewNames = Arrays.stream(viewNames).sorted().toArray(String[]::new);
+        for (String viewName : viewNames) {
+            catalog.createView(Identifier.create(databaseName, viewName), view, false);
+        }
+
+        // maxResults == null: return all views ordered by name, no next page.
+        pagedViews = catalog.listViewsPaged(databaseName, null, null, null);
+        assertThat(pagedViews.getElements()).containsExactly(sortedViewNames);
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        // maxResults == 0 is treated as "no paging" per the Catalog contract.
+        pagedViews = catalog.listViewsPaged(databaseName, 0, null, null);
+        assertThat(pagedViews.getElements()).containsExactly(sortedViewNames);
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        // An empty pattern means "no pattern": all views are returned.
+        pagedViews = catalog.listViewsPaged(databaseName, null, null, "");
+        assertThat(pagedViews.getElements()).containsExactly(sortedViewNames);
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        // Page through with maxResults = 2. nextPageToken is the last name of the current page.
+        int maxResults = 2;
+        pagedViews = catalog.listViewsPaged(databaseName, maxResults, null, null);
+        assertThat(pagedViews.getElements()).containsExactly("abd", "def");
+        assertThat(pagedViews.getNextPageToken()).isEqualTo("def");
+
+        pagedViews =
+                catalog.listViewsPaged(
+                        databaseName, maxResults, pagedViews.getNextPageToken(), null);
+        assertThat(pagedViews.getElements()).containsExactly("opr", "view1");
+        assertThat(pagedViews.getNextPageToken()).isEqualTo("view1");
+
+        pagedViews =
+                catalog.listViewsPaged(
+                        databaseName, maxResults, pagedViews.getNextPageToken(), null);
+        assertThat(pagedViews.getElements()).containsExactly("view2", "view3");
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        // maxResults larger than the full set returns everything in one page.
+        maxResults = 8;
+        pagedViews = catalog.listViewsPaged(databaseName, maxResults, null, null);
+        assertThat(pagedViews.getElements()).containsExactly(sortedViewNames);
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        // pageToken resumes strictly after the token value.
+        pagedViews = catalog.listViewsPaged(databaseName, maxResults, "view1", null);
+        assertThat(pagedViews.getElements()).containsExactly("view2", "view3");
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        // A negative maxResults is rejected.
+        assertThatThrownBy(() -> catalog.listViewsPaged(databaseName, -1, null, null))
+                .isInstanceOf(IllegalArgumentException.class);
+
+        // DatabaseNotExistException when the database does not exist.
+        final int finalMaxResults = maxResults;
+        assertThatThrownBy(
+                        () ->
+                                catalog.listViewsPaged(
+                                        "non_existing_db", finalMaxResults, "view1", null))
+                .isInstanceOf(Catalog.DatabaseNotExistException.class);
+
+        // Pattern filtering uses standard SQL LIKE: '%' matches any sequence, '_' a single char.
+        pagedViews = catalog.listViewsPaged(databaseName, null, null, "view%");
+        assertThat(pagedViews.getElements()).containsExactly("view1", "view2", "view3");
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        pagedViews = catalog.listViewsPaged(databaseName, null, null, "view_");
+        assertThat(pagedViews.getElements()).containsExactly("view1", "view2", "view3");
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        pagedViews = catalog.listViewsPaged(databaseName, null, null, "_bd");
+        assertThat(pagedViews.getElements()).containsExactly("abd");
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        pagedViews = catalog.listViewsPaged(databaseName, null, null, "zzz%");
+        assertThat(pagedViews.getElements()).isEmpty();
+        assertThat(pagedViews.getNextPageToken()).isNull();
+
+        // Pattern combined with paging.
+        pagedViews = catalog.listViewsPaged(databaseName, 2, null, "view%");
+        assertThat(pagedViews.getElements()).containsExactly("view1", "view2");
+        assertThat(pagedViews.getNextPageToken()).isEqualTo("view2");
+
+        pagedViews =
+                catalog.listViewsPaged(databaseName, 2, pagedViews.getNextPageToken(), "view%");
+        assertThat(pagedViews.getElements()).containsExactly("view3");
+        assertThat(pagedViews.getNextPageToken()).isNull();
+    }
+
+    @Override
+    @Test
+    public void testListViewDetailsPaged() throws Exception {
+        String databaseName = "view_details_paged_db";
+        catalog.createDatabase(databaseName, false);
+
+        PagedList<View> pagedViewDetails =
+                catalog.listViewDetailsPaged(databaseName, null, null, null);
+        assertThat(pagedViewDetails.getElements()).isEmpty();
+        assertThat(pagedViewDetails.getNextPageToken()).isNull();
+
+        View view = buildView(databaseName);
+        String[] viewNames = {"view1", "view2", "view3", "abd", "def", "opr"};
+        String[] sortedViewNames = Arrays.stream(viewNames).sorted().toArray(String[]::new);
+        for (String viewName : viewNames) {
+            catalog.createView(Identifier.create(databaseName, viewName), view, false);
+        }
+
+        pagedViewDetails = catalog.listViewDetailsPaged(databaseName, null, null, null);
+        assertThat(viewNames(pagedViewDetails)).containsExactly(sortedViewNames);
+        assertThat(pagedViewDetails.getNextPageToken()).isNull();
+
+        // maxResults == 0 and an empty pattern both mean "no filtering / no paging".
+        pagedViewDetails = catalog.listViewDetailsPaged(databaseName, 0, null, "");
+        assertThat(viewNames(pagedViewDetails)).containsExactly(sortedViewNames);
+        assertThat(pagedViewDetails.getNextPageToken()).isNull();
+
+        int maxResults = 2;
+        pagedViewDetails = catalog.listViewDetailsPaged(databaseName, maxResults, null, null);
+        assertThat(viewNames(pagedViewDetails)).containsExactly("abd", "def");
+        assertThat(pagedViewDetails.getNextPageToken()).isEqualTo("def");
+
+        pagedViewDetails =
+                catalog.listViewDetailsPaged(
+                        databaseName, maxResults, pagedViewDetails.getNextPageToken(), null);
+        assertThat(viewNames(pagedViewDetails)).containsExactly("opr", "view1");
+        assertThat(pagedViewDetails.getNextPageToken()).isEqualTo("view1");
+
+        pagedViewDetails =
+                catalog.listViewDetailsPaged(
+                        databaseName, maxResults, pagedViewDetails.getNextPageToken(), null);
+        assertThat(viewNames(pagedViewDetails)).containsExactly("view2", "view3");
+        assertThat(pagedViewDetails.getNextPageToken()).isNull();
+
+        // DatabaseNotExistException when the database does not exist.
+        final int finalMaxResults = maxResults;
+        assertThatThrownBy(
+                        () ->
+                                catalog.listViewDetailsPaged(
+                                        "non_existing_db", finalMaxResults, null, null))
+                .isInstanceOf(Catalog.DatabaseNotExistException.class);
+
+        // Pattern filtering.
+        pagedViewDetails = catalog.listViewDetailsPaged(databaseName, null, null, "view%");
+        assertThat(viewNames(pagedViewDetails)).containsExactly("view1", "view2", "view3");
+        assertThat(pagedViewDetails.getNextPageToken()).isNull();
+    }
+
+    private static List<String> viewNames(PagedList<View> views) {
+        return views.getElements().stream().map(View::name).collect(Collectors.toList());
+    }
+
     @Test
     public void testConcurrentCreateViewOnlyCreatesOneView() throws Exception {
         String databaseName = "concurrent_view_db";