HIVE-29878: Support pagination for list endpoints in HMS REST Catalog and remove redundant type casting in REST Catalog

standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogAdapter.java

@@ -37,6 +37,7 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.commons.lang3.math.NumberUtils;
 import org.apache.iceberg.catalog.ViewCatalog;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
@@ -89,6 +90,10 @@ public class HMSCatalogAdapter implements Closeable {
   private static final Logger LOG = LoggerFactory.getLogger(HMSCatalogAdapter.class);
   private static final Splitter SLASH = Splitter.on('/').omitEmptyStrings();
 
+  private static final String PAGE_TOKEN = "pageToken";
+  private static final String PAGE_SIZE = "pageSize";
+  private static final String PARENT = "parent";
+
   private static final String PREFIX_VAR = "prefix";
   private static final String PREFIX_PLACEHOLDER = "{" + PREFIX_VAR + "}";
 
@@ -275,23 +280,40 @@ public Class<? extends RESTRequest> requestClass() {
   private ConfigResponse config() {
     final List<Endpoint> endpoints = Arrays.stream(Route.values())
         .map(r -> Endpoint.create(r.method.name(), r.pathTemplate)).toList();
-    return castResponse(ConfigResponse.class, ConfigResponse.builder().withEndpoints(endpoints).build());
+    return ConfigResponse.builder().withEndpoints(endpoints).build();
   }
 
-  private ListNamespacesResponse listNamespaces(Map<String, String> vars) {
-    Namespace namespace;
-    if (vars.containsKey("parent")) {
-      namespace = Namespace.of(RESTUtil.NAMESPACE_SPLITTER.splitToStream(vars.get("parent")).toArray(String[]::new));
-    } else {
-      namespace = Namespace.empty();
+  /**
+   * Paging parameters of a list request. Without pageSize the listing is unpaged, expressed as a
+   * single unbounded first page. This relies on CatalogHandlers.paginate treating a null token as
+   * the first page and returning a null next-page-token once the list is exhausted.
+   */
+  private record PageRequest(String token, String size) {
+    private static final PageRequest UNPAGED =
+        new PageRequest(null, String.valueOf(Integer.MAX_VALUE));
+
+    static PageRequest from(Map<String, String> vars) {
+      String size = vars.get(PAGE_SIZE);
+      if (size == null) {
+        return UNPAGED; // token ignored, as upstream; keeping it would overflow token + MAX_VALUE
+      }
+      Preconditions.checkArgument(NumberUtils.toInt(size, 0) > 0,
+          "Invalid %s: %s, must be a positive integer", PAGE_SIZE, size);
+      return new PageRequest(vars.get(PAGE_TOKEN), size);
     }
-    return castResponse(ListNamespacesResponse.class, CatalogHandlers.listNamespaces(asNamespaceCatalog, namespace));
+  }
+
+  private ListNamespacesResponse listNamespaces(Map<String, String> vars) {
+    Namespace parent = vars.containsKey(PARENT)
+        ? RESTUtil.namespaceFromQueryParam(vars.get(PARENT))
+        : Namespace.empty();
+    PageRequest page = PageRequest.from(vars);
+    return CatalogHandlers.listNamespaces(asNamespaceCatalog, parent, page.token(), page.size());
   }
 
   private CreateNamespaceResponse createNamespace(Object body) {
     CreateNamespaceRequest request = castRequest(CreateNamespaceRequest.class, body);
-    return castResponse(
-        CreateNamespaceResponse.class, CatalogHandlers.createNamespace(asNamespaceCatalog, request));
+    return CatalogHandlers.createNamespace(asNamespaceCatalog, request);
   }
 
   private RESTResponse namespaceExists(Map<String, String> vars) {
@@ -302,8 +324,7 @@ private RESTResponse namespaceExists(Map<String, String> vars) {
 
   private GetNamespaceResponse loadNamespace(Map<String, String> vars) {
     Namespace namespace = namespaceFromPathVars(vars);
-    return castResponse(
-        GetNamespaceResponse.class, CatalogHandlers.loadNamespace(asNamespaceCatalog, namespace));
+    return CatalogHandlers.loadNamespace(asNamespaceCatalog, namespace);
   }
 
   private RESTResponse dropNamespace(Map<String, String> vars) {
@@ -315,29 +336,25 @@ private UpdateNamespacePropertiesResponse updateNamespace(Map<String, String> va
     Namespace namespace = namespaceFromPathVars(vars);
     UpdateNamespacePropertiesRequest request =
         castRequest(UpdateNamespacePropertiesRequest.class, body);
-    return castResponse(
-        UpdateNamespacePropertiesResponse.class,
-        CatalogHandlers.updateNamespaceProperties(asNamespaceCatalog, namespace, request));
+    return CatalogHandlers.updateNamespaceProperties(asNamespaceCatalog, namespace, request);
   }
 
   private ListTablesResponse listTables(Map<String, String> vars) {
     Namespace namespace = namespaceFromPathVars(vars);
-    return castResponse(ListTablesResponse.class, CatalogHandlers.listTables(catalog, namespace));
+    PageRequest page = PageRequest.from(vars);
+    return CatalogHandlers.listTables(catalog, namespace, page.token(), page.size());
   }
 
   private LoadTableResponse createTable(Map<String, String> vars, Object body) {
-    final Class<LoadTableResponse> responseType = LoadTableResponse.class;
     Namespace namespace = namespaceFromPathVars(vars);
     CreateTableRequest request = castRequest(CreateTableRequest.class, body);
     request.validate();
     if (request.stageCreate()) {
       Map<String, String> namespaceMetadata = asNamespaceCatalog.loadNamespaceMetadata(namespace);
       icebergAuthorizer.validateStageCreateTable(catalogName, namespace, namespaceMetadata, request);
-      return castResponse(
-              responseType, CatalogHandlers.stageTableCreate(catalog, namespace, request));
+      return CatalogHandlers.stageTableCreate(catalog, namespace, request);
     } else {
-      return castResponse(
-              responseType, CatalogHandlers.createTable(catalog, namespace, request));
+      return CatalogHandlers.createTable(catalog, namespace, request);
     }
   }
 
@@ -358,19 +375,19 @@ private RESTResponse tableExists(Map<String, String> vars) {
 
   private LoadTableResponse loadTable(Map<String, String> vars) {
     TableIdentifier ident = identFromPathVars(vars);
-    return castResponse(LoadTableResponse.class, CatalogHandlers.loadTable(catalog, ident));
+    return CatalogHandlers.loadTable(catalog, ident);
   }
 
   private LoadTableResponse registerTable(Map<String, String> vars, Object body) {
-      Namespace namespace = namespaceFromPathVars(vars);
-      RegisterTableRequest request = castRequest(RegisterTableRequest.class, body);
-      return castResponse(LoadTableResponse.class, CatalogHandlers.registerTable(catalog, namespace, request));
+    Namespace namespace = namespaceFromPathVars(vars);
+    RegisterTableRequest request = castRequest(RegisterTableRequest.class, body);
+    return CatalogHandlers.registerTable(catalog, namespace, request);
   }
 
   private LoadTableResponse updateTable(Map<String, String> vars, Object body) {
     TableIdentifier ident = identFromPathVars(vars);
     UpdateTableRequest request = castRequest(UpdateTableRequest.class, body);
-    return castResponse(LoadTableResponse.class, CatalogHandlers.updateTable(catalog, ident, request));
+    return CatalogHandlers.updateTable(catalog, ident, request);
   }
 
   private RESTResponse renameTable(Object body) {
@@ -395,23 +412,14 @@ private RESTResponse commitTransaction(Object body) {
 
   private ListTablesResponse listViews(Map<String, String> vars) {
     Namespace namespace = namespaceFromPathVars(vars);
-    String pageToken = PropertyUtil.propertyAsString(vars, "pageToken", null);
-    String pageSize = PropertyUtil.propertyAsString(vars, "pageSize", null);
-    if (pageSize != null) {
-      return castResponse(
-          ListTablesResponse.class,
-          CatalogHandlers.listViews(asViewCatalog, namespace, pageToken, pageSize));
-    } else {
-      return castResponse(
-          ListTablesResponse.class, CatalogHandlers.listViews(asViewCatalog, namespace));
-    }
+    PageRequest page = PageRequest.from(vars);
+    return CatalogHandlers.listViews(asViewCatalog, namespace, page.token(), page.size());
   }
 
   private LoadViewResponse createView(Map<String, String> vars, Object body) {
     Namespace namespace = namespaceFromPathVars(vars);
     CreateViewRequest request = castRequest(CreateViewRequest.class, body);
-    return castResponse(
-        LoadViewResponse.class, CatalogHandlers.createView(asViewCatalog, namespace, request));
+    return CatalogHandlers.createView(asViewCatalog, namespace, request);
   }
 
   private RESTResponse viewExists(Map<String, String> vars) {
@@ -422,14 +430,13 @@ private RESTResponse viewExists(Map<String, String> vars) {
 
   private LoadViewResponse loadView(Map<String, String> vars) {
     TableIdentifier ident = viewIdentFromPathVars(vars);
-    return castResponse(LoadViewResponse.class, CatalogHandlers.loadView(asViewCatalog, ident));
+    return CatalogHandlers.loadView(asViewCatalog, ident);
   }
 
   private LoadViewResponse updateView(Map<String, String> vars, Object body) {
     TableIdentifier ident = viewIdentFromPathVars(vars);
     UpdateTableRequest request = castRequest(UpdateTableRequest.class, body);
-    return castResponse(
-        LoadViewResponse.class, CatalogHandlers.updateView(asViewCatalog, ident, request));
+    return CatalogHandlers.updateView(asViewCatalog, ident, request);
   }
 
   private RESTResponse renameView(Object body) {
@@ -446,8 +453,7 @@ private RESTResponse dropView(Map<String, String> vars) {
   private LoadViewResponse registerView(Map<String, String> vars, Object body) {
     Namespace namespace = namespaceFromPathVars(vars);
     RegisterViewRequest request = castRequest(RegisterViewRequest.class, body);
-    return castResponse(
-        LoadViewResponse.class, CatalogHandlers.registerView(asViewCatalog, namespace, request));
+    return CatalogHandlers.registerView(asViewCatalog, namespace, request);
   }
 
   /**
@@ -554,12 +560,6 @@ public void close() {
     }
   }
 
-  private static class BadResponseType extends RuntimeException {
-    private BadResponseType(Class<?> responseType, Object response) {
-      super(
-          String.format("Invalid response object, not a %s: %s", responseType.getName(), response));
-    }
-  }
 
   private static class BadRequestType extends RuntimeException {
     private BadRequestType(Class<?> requestType, Object request) {
@@ -574,12 +574,6 @@ public static <T> T castRequest(Class<T> requestType, Object request) {
     throw new BadRequestType(requestType, request);
   }
 
-  public static <T extends RESTResponse> T castResponse(Class<T> responseType, Object response) {
-    if (responseType.isInstance(response)) {
-      return responseType.cast(response);
-    }
-    throw new BadResponseType(responseType, response);
-  }
 
   public static void configureResponseFromException(
       Exception exc, ErrorResponse.Builder errorBuilder) {
@@ -595,7 +589,7 @@ public static void configureResponseFromException(
   }
 
   private static Namespace namespaceFromPathVars(Map<String, String> pathVars) {
-    return RESTUtil.decodeNamespace(pathVars.get("namespace"));
+    return RESTUtil.decodeNamespace(pathVars.get("namespace"), "%1F");
   }
 
   private static TableIdentifier identFromPathVars(Map<String, String> pathVars) {

standalone-metastore/metastore-rest-catalog/src/test/java/org/apache/iceberg/rest/TestHMSCatalogAdapterPagination.java

@@ -0,0 +1,127 @@
+/*
+ * 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.iceberg.rest;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.ViewCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.rest.responses.ListNamespacesResponse;
+import org.apache.iceberg.rest.HTTPRequest.HTTPMethod;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import javax.servlet.http.HttpServletResponse;
+
+class TestHMSCatalogAdapterPagination {
+
+  private HMSCatalogAdapter adapter;
+  private HttpServletResponse response;
+  private StringWriter stringWriter;
+
+  @BeforeEach
+  void setup() throws Exception {
+    Catalog catalog =
+        Mockito.mock(
+            Catalog.class,
+            Mockito.withSettings().extraInterfaces(SupportsNamespaces.class, ViewCatalog.class));
+    SupportsNamespaces nsCatalog = (SupportsNamespaces) catalog;
+    List<Namespace> fakeNamespaces =
+        Arrays.asList(
+            Namespace.of("db1"),
+            Namespace.of("db2"),
+            Namespace.of("db3"),
+            Namespace.of("db4"),
+            Namespace.of("db5"));
+    Mockito.when(nsCatalog.listNamespaces()).thenReturn(fakeNamespaces);
+    Mockito.when(nsCatalog.listNamespaces(Mockito.any())).thenReturn(fakeNamespaces);
+
+    adapter = new HMSCatalogAdapter("test", catalog, null, Collections.emptyList());
+
+    response = Mockito.mock(HttpServletResponse.class);
+    stringWriter = new StringWriter();
+    Mockito.when(response.getWriter()).thenReturn(new PrintWriter(stringWriter));
+  }
+
+  @AfterEach
+  void tearDown() {
+    if (adapter != null) {
+      adapter.close();
+    }
+  }
+
+  @Test
+  void testUnpaginatedRequest() throws Exception {
+    // Missing pageSize (should call unpaginated and succeed without NumberFormatException)
+    Map<String, String> vars = ImmutableMap.of("pageToken", "0");
+    ListNamespacesResponse res =
+        adapter.execute(HTTPMethod.GET, "v1/namespaces", vars, null, response);
+
+    if (res == null) {
+      System.err.println("Error output: " + stringWriter);
+    }
+    Assertions.assertNotNull(res, "Response should not be null");
+    Assertions.assertEquals(5, res.namespaces().size(), "Should return all 5 unpaginated");
+  }
+
+  @Test
+  void testPaginatedRequest() throws Exception {
+    // Both pageToken and pageSize (should call paginated and slice without errors)
+    Map<String, String> vars = ImmutableMap.of("pageToken", "0", "pageSize", "2");
+    ListNamespacesResponse res =
+        adapter.execute(HTTPMethod.GET, "v1/namespaces", vars, null, response);
+
+    Assertions.assertNotNull(res, "Response should not be null");
+    Assertions.assertEquals(2, res.namespaces().size(), "Should return exactly 2 paginated items");
+  }
+
+  @Test
+  void testPaginatedRequestWithoutToken() throws Exception {
+    // pageSize without pageToken
+    Map<String, String> vars = ImmutableMap.of("pageSize", "2");
+    ListNamespacesResponse res =
+        adapter.execute(HTTPMethod.GET, "v1/namespaces", vars, null, response);
+
+    Assertions.assertNotNull(res, "Response should not be null");
+    Assertions.assertEquals(2, res.namespaces().size(), "Should return exactly 2 paginated items");
+  }
+
+  @Test
+  void testInvalidPageSize() throws Exception {
+    // Invalid pageSize (should return null and write error to response)
+    Map<String, String> vars = ImmutableMap.of("pageSize", "invalid");
+    ListNamespacesResponse res =
+        adapter.execute(HTTPMethod.GET, "v1/namespaces", vars, null, response);
+
+    Assertions.assertNull(res, "Response should be null because it encountered an error");
+    Assertions.assertTrue(
+        stringWriter.toString().contains("must be a positive integer"),
+        "Error output should contain the Preconditions error message");
+  }
+}