Core: Use HEAD request to check if namespace exists

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

@@ -163,6 +163,12 @@ public static CreateNamespaceResponse createNamespace(
         .build();
   }
 
+  public static void namespaceExists(SupportsNamespaces catalog, Namespace namespace) {
+    if (!catalog.namespaceExists(namespace)) {
+      throw new NoSuchNamespaceException("Namespace does not exist: %s", namespace);
+    }
+  }
+
   public static GetNamespaceResponse loadNamespace(
       SupportsNamespaces catalog, Namespace namespace) {
     Map<String, String> properties = catalog.loadNamespaceMetadata(namespace);

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

@@ -228,6 +228,11 @@ public List<Namespace> listNamespaces(Namespace ns) throws NoSuchNamespaceExcept
     return nsDelegate.listNamespaces(ns);
   }
 
+  @Override
+  public boolean namespaceExists(Namespace namespace) {
+    return nsDelegate.namespaceExists(namespace);
+  }
+
   @Override
   public Map<String, String> loadNamespaceMetadata(Namespace ns) throws NoSuchNamespaceException {
     return nsDelegate.loadNamespaceMetadata(ns);

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

@@ -652,6 +652,19 @@ public List<Namespace> listNamespaces(SessionContext context, Namespace namespac
     return namespaces.build();
   }
 
+  @Override
+  public boolean namespaceExists(SessionContext context, Namespace namespace) {
+    checkNamespaceIsValid(namespace);
+
+    try {
+      client.head(
+          paths.namespace(namespace), headers(context), ErrorHandlers.namespaceErrorHandler());
+      return true;
+    } catch (NoSuchNamespaceException e) {
+      return false;
+    }
+  }
+
   @Override
   public Map<String, String> loadNamespaceMetadata(SessionContext context, Namespace ns) {
     Endpoint.check(endpoints, Endpoint.V1_LOAD_NAMESPACE);

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

@@ -124,6 +124,7 @@ enum Route {
         ResourcePaths.V1_NAMESPACES,
         CreateNamespaceRequest.class,
         CreateNamespaceResponse.class),
+    NAMESPACE_EXISTS(HTTPMethod.HEAD, ResourcePaths.V1_NAMESPACE),
     LOAD_NAMESPACE(HTTPMethod.GET, ResourcePaths.V1_NAMESPACE, null, GetNamespaceResponse.class),
     DROP_NAMESPACE(HTTPMethod.DELETE, ResourcePaths.V1_NAMESPACE),
     UPDATE_NAMESPACE(
@@ -330,6 +331,13 @@ public <T extends RESTResponse> T handleRequest(
         }
         break;
 
+      case NAMESPACE_EXISTS:
+        if (asNamespaceCatalog != null) {
+          CatalogHandlers.namespaceExists(asNamespaceCatalog, namespaceFromPathVars(vars));
+          return null;
+        }
+        break;
+
       case LOAD_NAMESPACE:
         if (asNamespaceCatalog != null) {
           Namespace namespace = namespaceFromPathVars(vars);

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

@@ -2634,6 +2634,35 @@ public void testNoCleanupForNonCleanableReplaceTransaction() {
         .isTrue();
   }
 
+  @Test
+  public void testNamespaceExistsViaHEADRequest() {
+    RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
+    RESTCatalog catalog =
+        new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
+    catalog.initialize("test", ImmutableMap.of());
+
+    assertThat(catalog.namespaceExists(Namespace.of("non-existing"))).isFalse();
+
+    Mockito.verify(adapter)
+        .execute(
+            eq(HTTPMethod.GET),
+            eq("v1/config"),
+            any(),
+            any(),
+            eq(ConfigResponse.class),
+            any(),
+            any());
+    Mockito.verify(adapter)
+        .execute(
+            eq(HTTPMethod.HEAD),
+            eq("v1/namespaces/non-existing"),
+            any(),
+            any(),
+            any(),
+            any(),
+            any());
+  }
+
   private RESTCatalog catalog(RESTCatalogAdapter adapter) {
     RESTCatalog catalog =
         new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);