Report all missing namespaces at once (#7671)

api/NESSIE-SPEC-2-0.md

@@ -22,6 +22,13 @@ well-specified behaviours.
 
 Refer to the [Nessie API documentation](./README.md) for the meaning of Nessie-specific terms.
 
+# 2.1.3
+
+* Released with Nessie version 0.73.0.
+* When a commit attempts to create a content inside a non-existing namespace, the server will not 
+  only return a `NAMESPACE_ABSENT` conflict for the non-existing namespace itself, but will also 
+  return additional `NAMESPACE_ABSENT` conflicts for all the non-existing ancestor namespaces.
+
 # 2.1.2
 
 * Released with Nessie version 0.68.0.

versioned/storage/store/src/main/java/org/projectnessie/versioned/storage/versionstore/BaseCommitHelper.java

@@ -357,13 +357,19 @@ private void validateNamespacesExistForContentKeys(
       }
 
       StoreIndexElement<CommitOp> ns = headIndex.get(keyToStoreKey(key));
-      if (ns == null || !ns.content().action().exists()) {
+      for (;
+          ns == null || !ns.content().action().exists();
+          key = key.getParent(), ns = headIndex.get(keyToStoreKey(key))) {
         conflictConsumer.accept(
             conflict(
                 Conflict.ConflictType.NAMESPACE_ABSENT,
                 key,
                 format("namespace '%s' must exist", key)));
-      } else {
+        if (key.getElementCount() == 1) {
+          break;
+        }
+      }
+      if (ns != null && ns.content().action().exists()) {
         CommitOp nsContent = ns.content();
         if (nsContent.payload() != payloadForContent(Content.Type.NAMESPACE)) {
           conflictConsumer.accept(

versioned/tests/src/main/java/org/projectnessie/versioned/tests/AbstractNamespaceValidation.java

@@ -21,6 +21,7 @@
 import static org.assertj.core.api.Assertions.tuple;
 import static org.assertj.core.api.InstanceOfAssertFactories.list;
 import static org.assertj.core.api.InstanceOfAssertFactories.type;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 import static org.projectnessie.model.CommitMeta.fromMessage;
 import static org.projectnessie.model.Conflict.ConflictType.NAMESPACE_ABSENT;
 import static org.projectnessie.model.Conflict.ConflictType.NOT_A_NAMESPACE;
@@ -62,16 +63,17 @@ protected AbstractNamespaceValidation(VersionStore store) {
   }
 
   static Stream<ContentKey> contentKeys() {
-    return Stream.of(
-        ContentKey.of("ns", "table"),
-        ContentKey.of("ns", "table"),
-        ContentKey.of("ns2", "ns", "table"),
-        ContentKey.of("ns2", "ns", "table"));
+    return Stream.of(ContentKey.of("ns", "table"), ContentKey.of("ns2", "ns", "table"));
   }
 
   @ParameterizedTest
   @MethodSource("contentKeys")
   void commitWithNonExistingNamespace(ContentKey key) throws Exception {
+
+    assumeTrue(
+        key.getElementCount() == 2 || !isNewStorageModel(),
+        "multiple missing namespaces are tested separately below for the new storage");
+
     BranchName branch = BranchName.of("commitWithNonExistingNamespace");
     store().create(branch, Optional.empty());
 
@@ -123,6 +125,106 @@ void commitWithNonExistingNamespace(ContentKey key) throws Exception {
                 "namespace '" + key.getNamespace() + "' must exist"));
   }
 
+  @Test
+  void commitWithNonExistingNamespaceMultipleNewStorage() throws Exception {
+    assumeTrue(isNewStorageModel());
+    ContentKey a = ContentKey.of("a");
+    ContentKey b = ContentKey.of("a", "b");
+    ContentKey c = ContentKey.of("a", "b", "c");
+    ContentKey table = ContentKey.of("a", "b", "c", "table");
+    BranchName branch = BranchName.of("commitWithNonExistingNamespace");
+    store().create(branch, Optional.empty());
+
+    store()
+        .commit(
+            branch,
+            Optional.empty(),
+            fromMessage("create ns a"),
+            singletonList(Put.of(a, Namespace.of(a))));
+
+    soft.assertThatThrownBy(
+            () ->
+                store()
+                    .commit(
+                        branch,
+                        Optional.empty(),
+                        fromMessage("non-existing-ns"),
+                        singletonList(Put.of(table, newOnRef("value")))))
+        .isInstanceOf(ReferenceConflictException.class)
+        .hasMessage(
+            "There are multiple conflicts that prevent committing the provided operations: "
+                + "namespace 'a.b.c' must exist, "
+                + "namespace 'a.b' must exist.")
+        .asInstanceOf(type(ReferenceConflictException.class))
+        .extracting(ReferenceConflictException::getReferenceConflicts)
+        .extracting(ReferenceConflicts::conflicts, list(Conflict.class))
+        .extracting(Conflict::conflictType, Conflict::key, Conflict::message)
+        .containsExactly(
+            tuple(NAMESPACE_ABSENT, c, "namespace 'a.b.c' must exist"),
+            tuple(NAMESPACE_ABSENT, b, "namespace 'a.b' must exist"));
+
+    store()
+        .commit(
+            branch,
+            Optional.empty(),
+            fromMessage("create content b"),
+            singletonList(Put.of(b, newOnRef("value"))));
+
+    soft.assertThatThrownBy(
+            () ->
+                store()
+                    .commit(
+                        branch,
+                        Optional.empty(),
+                        fromMessage("non-existing-ns"),
+                        singletonList(Put.of(table, newOnRef("value")))))
+        .isInstanceOf(ReferenceConflictException.class)
+        .hasMessage(
+            "There are multiple conflicts that prevent committing the provided operations: "
+                + "namespace 'a.b.c' must exist, "
+                + "expecting the key 'a.b' to be a namespace, but is not a namespace (using a content object that is not a namespace as a namespace is forbidden).")
+        .asInstanceOf(type(ReferenceConflictException.class))
+        .extracting(ReferenceConflictException::getReferenceConflicts)
+        .extracting(ReferenceConflicts::conflicts, list(Conflict.class))
+        .extracting(Conflict::conflictType, Conflict::key, Conflict::message)
+        .containsExactly(
+            tuple(NAMESPACE_ABSENT, c, "namespace 'a.b.c' must exist"),
+            tuple(
+                NOT_A_NAMESPACE,
+                b,
+                "expecting the key 'a.b' to be a namespace, but is not a namespace (using a content object that is not a namespace as a namespace is forbidden)"));
+
+    store()
+        .commit(
+            branch, Optional.empty(), fromMessage("delete content b"), singletonList(Delete.of(b)));
+    store()
+        .commit(
+            branch, Optional.empty(), fromMessage("delete content a"), singletonList(Delete.of(a)));
+
+    soft.assertThatThrownBy(
+            () ->
+                store()
+                    .commit(
+                        branch,
+                        Optional.empty(),
+                        fromMessage("non-existing-ns"),
+                        singletonList(Put.of(table, newOnRef("value")))))
+        .isInstanceOf(ReferenceConflictException.class)
+        .hasMessage(
+            "There are multiple conflicts that prevent committing the provided operations: "
+                + "namespace 'a.b.c' must exist, "
+                + "namespace 'a.b' must exist, "
+                + "namespace 'a' must exist.")
+        .asInstanceOf(type(ReferenceConflictException.class))
+        .extracting(ReferenceConflictException::getReferenceConflicts)
+        .extracting(ReferenceConflicts::conflicts, list(Conflict.class))
+        .extracting(Conflict::conflictType, Conflict::key, Conflict::message)
+        .containsExactly(
+            tuple(NAMESPACE_ABSENT, c, "namespace 'a.b.c' must exist"),
+            tuple(NAMESPACE_ABSENT, b, "namespace 'a.b' must exist"),
+            tuple(NAMESPACE_ABSENT, a, "namespace 'a' must exist"));
+  }
+
   @ParameterizedTest
   @MethodSource("contentKeys")
   void commitWithNonNamespace(ContentKey key) throws Exception {