Core, API: Add support for recursive namespace listing

api/src/main/java/org/apache/iceberg/catalog/SupportsNamespaces.java

@@ -22,8 +22,10 @@
 import java.util.Map;
 import java.util.Set;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 
 /**
@@ -74,6 +76,28 @@ default List<Namespace> listNamespaces() {
     return listNamespaces(Namespace.empty());
   }
 
+  /**
+   * List namespaces from the catalog recursively.
+   *
+   * <p>If an object such as a table, view, or function exists, its parent namespaces must also
+   * exist and must be returned by this discovery method. For example, if table a.b.t exists, this
+   * method must return ["a"] in the result array.
+   *
+   * @return a List of namespace {@link Namespace} names
+   */
+  default List<Namespace> listNamespacesRecursively() {
+    ImmutableList.Builder<Namespace> namespaces = ImmutableList.builder();
+    try {
+      for (Namespace ns : listNamespaces(Namespace.empty())) {
+        namespaces.add(ns);
+        namespaces.addAll(listNamespacesRecursively(ns));
+      }
+    } catch (NoSuchNamespaceException | ForbiddenException e) {
+      // Skip the namespace if it has been deleted concurrently or access is forbidden
+    }
+    return namespaces.build();
+  }
+
   /**
    * List child namespaces from the namespace.
    *
@@ -104,6 +128,48 @@ default List<Namespace> listNamespaces() {
    */
   List<Namespace> listNamespaces(Namespace namespace) throws NoSuchNamespaceException;
 
+  /**
+   * List child namespaces from the namespace recursively.
+   *
+   * <p>For two existing tables named 'a.b.c.table' and 'a.b.d.table', this method returns:
+   *
+   * <ul>
+   *   <li>Given: {@code Namespace.empty()}
+   *   <li>Returns: {@code Namespace.of("a")}
+   * </ul>
+   *
+   * <ul>
+   *   <li>Given: {@code Namespace.of("a")}
+   *   <li>Returns: {@code Namespace.of("a", "b")}
+   * </ul>
+   *
+   * <ul>
+   *   <li>Given: {@code Namespace.of("a", "b")}
+   *   <li>Returns: {@code Namespace.of("a", "b", "c")} and {@code Namespace.of("a", "b", "d")}
+   * </ul>
+   *
+   * <ul>
+   *   <li>Given: {@code Namespace.of("a", "b", "c")}
+   *   <li>Returns: empty list, because there are no child namespaces
+   * </ul>
+   *
+   * @return a List of child {@link Namespace} names from the given namespace
+   * @throws NoSuchNamespaceException If the namespace does not exist (optional)
+   */
+  default List<Namespace> listNamespacesRecursively(Namespace namespace)
+      throws NoSuchNamespaceException {
+    ImmutableList.Builder<Namespace> namespaces = ImmutableList.builder();
+    try {
+      for (Namespace ns : listNamespaces(namespace)) {
+        namespaces.add(ns);
+        namespaces.addAll(listNamespacesRecursively(ns));
+      }
+    } catch (NoSuchNamespaceException | ForbiddenException e) {
+      // Skip the namespace if it has been deleted concurrently or access is forbidden
+    }
+    return namespaces.build();
+  }
+
   /**
    * Load metadata properties for a namespace.
    *

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

@@ -70,13 +70,15 @@
 import org.apache.iceberg.catalog.ViewCatalog;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.exceptions.NoSuchViewException;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.rest.RESTCatalogProperties.SnapshotMode;
@@ -302,16 +304,48 @@ private static <T> Pair<List<T>, String> paginate(List<T> list, String pageToken
 
   public static ListNamespacesResponse listNamespaces(
       SupportsNamespaces catalog, Namespace parent) {
+    return listNamespaces(catalog, parent, false);
+  }
+
+  public static ListNamespacesResponse listNamespacesRecursively(
+      SupportsNamespaces catalog, Namespace parent) {
+    return listNamespaces(catalog, parent, true);
+  }
+
+  private static ListNamespacesResponse listNamespaces(
+      SupportsNamespaces catalog, Namespace parent, boolean recursive) {
     List<Namespace> results;
     if (parent.isEmpty()) {
       results = catalog.listNamespaces();
     } else {
       results = catalog.listNamespaces(parent);
     }
 
+    if (recursive) {
+      results = listNamespacesRecursively(catalog, results);
+    }
+
     return ListNamespacesResponse.builder().addAll(results).build();
   }
 
+  private static List<Namespace> listNamespacesRecursively(
+      SupportsNamespaces catalog, List<Namespace> namespaces) {
+    List<Namespace> allNamespaces = Lists.newArrayList(namespaces);
+
+    for (Namespace namespace : namespaces) {
+      try {
+        List<Namespace> children = catalog.listNamespaces(namespace);
+        if (!children.isEmpty()) {
+          allNamespaces.addAll(listNamespacesRecursively(catalog, children));
+        }
+      } catch (NoSuchNamespaceException | ForbiddenException e) {
+        // Skip the namespace if it has been deleted concurrently or access is forbidden
+      }
+    }
+
+    return allNamespaces;
+  }
+
   public static ListNamespacesResponse listNamespaces(
       SupportsNamespaces catalog, Namespace parent, String pageToken, String pageSize) {
     List<Namespace> results;

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -751,6 +751,15 @@ public void createNamespace(
 
   @Override
   public List<Namespace> listNamespaces(SessionContext context, Namespace namespace) {
+    return listNamespaces(context, namespace, false);
+  }
+
+  public List<Namespace> listNamespacesRecursively(SessionContext context, Namespace namespace) {
+    return listNamespaces(context, namespace, true);
+  }
+
+  private List<Namespace> listNamespaces(
+      SessionContext context, Namespace namespace, boolean recursive) {
     if (!endpoints.contains(Endpoint.V1_LIST_NAMESPACES)) {
       return ImmutableList.of();
     }
@@ -760,6 +769,10 @@ public List<Namespace> listNamespaces(SessionContext context, Namespace namespac
       queryParams.put("parent", RESTUtil.namespaceToQueryParam(namespace, namespaceSeparator));
     }
 
+    if (recursive) {
+      queryParams.put("recursive", "true");
+    }
+
     ImmutableList.Builder<Namespace> namespaces = ImmutableList.builder();
     String pageToken = "";
     if (pageSize != null) {

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

@@ -190,6 +190,7 @@ public <T extends RESTResponse> T handleRequest(
           } else {
             ns = Namespace.empty();
           }
+          boolean recursive = PropertyUtil.propertyAsBoolean(vars, "recursive", false);
 
           String pageToken = PropertyUtil.propertyAsString(vars, "pageToken", null);
           String pageSize = PropertyUtil.propertyAsString(vars, "pageSize", null);
@@ -199,6 +200,10 @@ public <T extends RESTResponse> T handleRequest(
                 responseType,
                 CatalogHandlers.listNamespaces(asNamespaceCatalog, ns, pageToken, pageSize));
           } else {
+            if (recursive) {
+              return castResponse(
+                  responseType, CatalogHandlers.listNamespacesRecursively(asNamespaceCatalog, ns));
+            }
             return castResponse(
                 responseType, CatalogHandlers.listNamespaces(asNamespaceCatalog, ns));
           }

core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java

@@ -3549,6 +3549,60 @@ private void runConfigurableNamespaceSeparatorTest(
             any());
   }
 
+  @Test
+  public void testListNamespacesRecursively() throws IOException {
+    InMemoryCatalog backend = new InMemoryCatalog();
+    backend.initialize(
+        "backend", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, temp.toString()));
+    RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backend));
+    RESTCatalog catalog = catalog(adapter);
+
+    try {
+      // Create a namespace hierarchy:
+      // - ns1
+      //   - ns1.child1
+      //   - ns1.child2
+      //     - ns1.child2.grandchild
+      // - ns2
+      Namespace ns1 = Namespace.of("ns1");
+      Namespace ns1Child1 = Namespace.of("ns1", "child1");
+      Namespace ns1Child2 = Namespace.of("ns1", "child2");
+      Namespace ns1Child2Grandchild = Namespace.of("ns1", "child2", "grandchild");
+      Namespace ns2 = Namespace.of("ns2");
+
+      catalog.createNamespace(ns1);
+      catalog.createNamespace(ns1Child1);
+      catalog.createNamespace(ns1Child2);
+      catalog.createNamespace(ns1Child2Grandchild);
+      catalog.createNamespace(ns2);
+
+      // recursive listing - should get all namespaces in the hierarchy
+      assertThat(catalog.listNamespacesRecursively())
+          .containsExactlyInAnyOrder(ns1, ns1Child1, ns1Child2, ns1Child2Grandchild, ns2);
+
+      // non-recursive listing - should only get direct children
+      assertThat(catalog.listNamespaces()).containsExactlyInAnyOrder(ns1, ns2);
+
+      // recursive listing of ns1 - should get all descendants of ns1
+      assertThat(catalog.listNamespacesRecursively(ns1))
+          .containsExactlyInAnyOrder(ns1Child1, ns1Child2, ns1Child2Grandchild);
+
+      // non-recursive listing of ns1 - should only get direct children
+      assertThat(catalog.listNamespaces(ns1)).containsExactlyInAnyOrder(ns1Child1, ns1Child2);
+
+      // recursive listing of ns1.child2 - should only get direct children because
+      // nested namespaces don't exist
+      assertThat(catalog.listNamespacesRecursively(ns1Child2)).containsExactly(ns1Child2Grandchild);
+
+      // non-recursive listing of ns1.child2 - should only get direct children
+      assertThat(catalog.listNamespaces(ns1Child2)).containsExactly(ns1Child2Grandchild);
+      assertThat(catalog.listNamespacesRecursively(ns1Child2)).containsExactly(ns1Child2Grandchild);
+    } finally {
+      catalog.close();
+      backend.close();
+    }
+  }
+
   private RESTCatalog createCatalogWithIdempAdapter(ConfigResponse cfg, boolean expectOnMutations) {
     RESTCatalogAdapter adapter =
         Mockito.spy(

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

@@ -280,6 +280,13 @@ paths:
           schema:
             type: string
           example: "accounting%1Ftax"
+        - name: recursive
+          in: query
+          description: An optional flag indicating whether to list namespaces recursively.
+          required: false
+          schema:
+            type: boolean
+            default: false
       responses:
         200:
           $ref: '#/components/responses/ListNamespacesResponse'