Support create multiple element ns together for nessie

nessie/src/main/java/org/apache/iceberg/nessie/NessieIcebergClient.java

@@ -43,6 +43,8 @@
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.exceptions.NoSuchViewException;
 import org.apache.iceberg.relocated.com.google.common.base.Suppliers;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.Tasks;
 import org.apache.iceberg.view.ViewMetadata;
@@ -219,15 +221,41 @@ public void createNamespace(Namespace namespace, Map<String, String> metadata) {
     checkNamespaceIsValid(namespace);
     getRef().checkMutable();
     ContentKey key = ContentKey.of(namespace.levels());
-    org.projectnessie.model.Namespace content =
-        org.projectnessie.model.Namespace.of(key.getElements(), metadata);
     try {
       Content existing = api.getContent().reference(getReference()).key(key).get().get(key);
       if (existing != null) {
         throw namespaceAlreadyExists(key, existing, null);
       }
+
+      // check if the parent namespace exists
+      List<ContentKey> keys = Lists.newArrayList();
+      Map<ContentKey, Content> contentInfo = Maps.newHashMap();
+      for (int i = 0; i < namespace.levels().length; i++) {
+        Namespace parent = Namespace.of(Arrays.copyOf(namespace.levels(), i + 1));
+        ContentKey parentKey = ContentKey.of(parent.levels());
+        keys.add(parentKey);
+        Content parentContent =
+            api.getContent().reference(getReference()).key(parentKey).get().get(parentKey);
+        contentInfo.put(parentKey, parentContent);
+      }
+
+      List<Operation.Put> putOperations = Lists.newArrayList();
+      for (Map.Entry<ContentKey, Content> contentKeyContentEntry : contentInfo.entrySet()) {
+        if (contentKeyContentEntry.getValue() == null) {
+          org.projectnessie.model.Namespace content;
+          if (contentKeyContentEntry.getKey().equals(key)) {
+            content = org.projectnessie.model.Namespace.of(key.getElements(), metadata);
+          } else {
+            content =
+                org.projectnessie.model.Namespace.of(
+                    contentKeyContentEntry.getKey().getElements(), ImmutableMap.of());
+          }
+          putOperations.add(Operation.Put.of(contentKeyContentEntry.getKey(), content));
+        }
+      }
+
       try {
-        commitRetry("create namespace " + key, Operation.Put.of(key, content));
+        commitRetry("create namespace " + keys, putOperations.toArray(new Operation.Put[0]));
       } catch (NessieReferenceConflictException e) {
         Optional<Conflict> conflict =
             NessieUtil.extractSingleConflict(

nessie/src/test/java/org/apache/iceberg/nessie/TestMultipleClients.java

@@ -71,6 +71,18 @@ public void afterEach() throws Exception {
     anotherCatalog.close();
   }
 
+  @Test
+  public void testCreateNamespacesWithLevels() {
+    catalog.createNamespace(Namespace.of("l1", "l2", "l3", "l4"));
+    assertThat(catalog.listNamespaces()).containsExactlyInAnyOrder(Namespace.of("l1"));
+    assertThat(catalog.listNamespaces(Namespace.of("l1")))
+        .containsExactlyInAnyOrder(Namespace.of("l1", "l2"));
+    assertThat(catalog.listNamespaces(Namespace.of("l1", "l2")))
+        .containsExactlyInAnyOrder(Namespace.of("l1", "l2", "l3"));
+    assertThat(catalog.listNamespaces(Namespace.of("l1", "l2", "l3")))
+        .containsExactlyInAnyOrder(Namespace.of("l1", "l2", "l3", "l4"));
+  }
+
   @Test
   public void testListNamespaces() throws NessieConflictException, NessieNotFoundException {
     assertThat(catalog.listNamespaces()).isEmpty();

nessie/src/test/java/org/apache/iceberg/nessie/TestNessieIcebergClient.java

@@ -132,13 +132,33 @@ public void testCreateNamespace() throws NessieConflictException, NessieNotFound
         .extracting(LogResponse.LogEntry::getCommitMeta)
         .extracting(CommitMeta::getMessage, CommitMeta::getAuthor, CommitMeta::getProperties)
         .containsExactly(
-            "create namespace a",
+            "create namespace [a]",
             "iceberg-user",
             ImmutableMap.of(
                 "application-type", "iceberg",
                 "app-id", "iceberg-nessie"));
   }
 
+  @Test
+  public void testCreateNamespaceWithMultipleLevels()
+      throws NessieConflictException, NessieNotFoundException {
+    String branch = "createNamespaceWithMultipleLevelsBranch";
+    createBranch(branch);
+    Map<String, String> catalogOptions =
+        Map.of(
+            CatalogProperties.USER, "iceberg-user",
+            CatalogProperties.APP_ID, "iceberg-nessie");
+    NessieIcebergClient client = new NessieIcebergClient(api, branch, null, catalogOptions);
+
+    client.createNamespace(Namespace.of("a", "b", "c"), Map.of());
+    assertThat(client.listNamespaces(Namespace.of("a"))).isNotNull();
+    assertThat(client.listNamespaces(Namespace.of("a", "b"))).isNotNull();
+    assertThat(client.listNamespaces(Namespace.of("a", "b", "c"))).isNotNull();
+
+    client.createNamespace(Namespace.of("a", "b", "d"), Map.of());
+    assertThat(client.listNamespaces(Namespace.of("a", "b", "d"))).isNotNull();
+  }
+
   @Test
   public void testCreateNamespaceInvalid() throws NessieConflictException, NessieNotFoundException {
     String branch = "createNamespaceInvalidBranch";
@@ -148,10 +168,6 @@ public void testCreateNamespaceInvalid() throws NessieConflictException, NessieN
     assertThatThrownBy(() -> client.createNamespace(Namespace.empty(), Map.of()))
         .isInstanceOf(NoSuchNamespaceException.class)
         .hasMessageContaining("Invalid namespace: ");
-
-    assertThatThrownBy(() -> client.createNamespace(Namespace.of("a", "b"), Map.of()))
-        .isInstanceOf(NoSuchNamespaceException.class)
-        .hasMessageContaining("Cannot create namespace 'a.b': parent namespace 'a' does not exist");
   }
 
   @Test