Fix behavior when swapping tables (#7498)

A DDL sequence like
```sql
CREATE TABLE t ...;
CREATE TABLE t_tmp ...;
ALTER TABLE t RENAME TO t_bak;
ALTER TABLE t_tmp RENAME TO t;
```
yields to the wrong error `New value for key 't' must not have a content ID`.

The root-cause is an incomplete check in `CommitImpl.commitAddPut()` that does not take the `CommitOp.Action.exists()` flag into account, letting the "new content must not have a content ID"-assertion firing.

Fixes #7492

CHANGELOG.md

@@ -19,6 +19,7 @@ as necessary. Empty sections will not end in the release notes.
 ### Deprecations
 
 ### Fixes
+- Fix wrong `New value for key 'some-key' must not have a content ID` when swapping tables.
 
 ### Commits
 

servers/jax-rs-tests/src/main/java/org/projectnessie/jaxrs/tests/BaseTestNessieApi.java

@@ -32,6 +32,7 @@
 import static org.projectnessie.model.CommitMeta.fromMessage;
 import static org.projectnessie.model.FetchOption.ALL;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.hash.Hashing;
 import java.time.Duration;
@@ -1943,4 +1944,79 @@ void invalidCreateRepositoryConfig() throws Exception {
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessageContaining("Failed to parse default-cutoff-value");
   }
+
+  @Test
+  @NessieApiVersions(versions = NessieApiVersion.V2)
+  void renameTwice() throws Exception {
+    Branch main = api().getDefaultBranch();
+    soft.assertThat(api().getAllReferences().get().getReferences()).containsExactly(main);
+
+    ContentKey key = ContentKey.of("table");
+    ContentKey keyBackup = ContentKey.of("table_backup");
+    ContentKey keyTemp = ContentKey.of("table_tmp");
+    List<ContentKey> keys = ImmutableList.of(key, keyTemp, keyBackup);
+
+    IcebergTable tableOld = IcebergTable.of("old", 1, 2, 3, 4);
+    IcebergTable tableNew = IcebergTable.of("new", 1, 2, 3, 4);
+
+    CommitResponse createTable =
+        prepCommit(main, "commit", Put.of(key, tableOld)).commitWithResponse();
+    tableOld = createTable.contentWithAssignedId(key, tableOld);
+
+    soft.assertThat(api().getContent().reference(createTable.getTargetBranch()).key(key).get())
+        .hasSize(1)
+        .containsEntry(key, tableOld);
+
+    // create new "table_tmp"
+
+    CommitResponse createNewTemp =
+        prepCommit(createTable.getTargetBranch(), "new", Put.of(keyTemp, tableNew))
+            .commitWithResponse();
+    tableNew = createNewTemp.contentWithAssignedId(keyTemp, tableNew);
+
+    soft.assertThat(api().getContent().reference(createNewTemp.getTargetBranch()).keys(keys).get())
+        .hasSize(2)
+        .containsEntry(key, tableOld)
+        .containsEntry(keyTemp, tableNew);
+
+    // rename "original" to "original_backup"
+
+    CommitResponse renameToBackup =
+        prepCommit(
+                createNewTemp.getTargetBranch(),
+                "backup",
+                Delete.of(key),
+                Put.of(keyBackup, tableOld))
+            .commitWithResponse();
+
+    soft.assertThat(api().getContent().reference(renameToBackup.getTargetBranch()).keys(keys).get())
+        .hasSize(2)
+        .containsEntry(keyBackup, tableOld)
+        .containsEntry(keyTemp, tableNew);
+
+    // rename new "table_tmp" to "table"
+
+    CommitResponse renameNew =
+        prepCommit(
+                renameToBackup.getTargetBranch(),
+                "rename new",
+                Delete.of(keyTemp),
+                Put.of(key, tableNew))
+            .commitWithResponse();
+
+    soft.assertThat(api().getContent().reference(renameNew.getTargetBranch()).keys(keys).get())
+        .hasSize(2)
+        .containsEntry(keyBackup, tableOld)
+        .containsEntry(key, tableNew);
+
+    // delete backup "table_backup"
+
+    CommitResponse deleteOld =
+        prepCommit(renameNew.getTargetBranch(), "delete", Delete.of(keyBackup))
+            .commitWithResponse();
+
+    soft.assertThat(api().getContent().reference(deleteOld.getTargetBranch()).keys(keys).get())
+        .hasSize(1)
+        .containsEntry(key, tableNew);
+  }
 }

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

@@ -412,12 +412,13 @@ private void commitAddPut(
     StoreIndexElement<CommitOp> existing = expectedIndex.get(storeKey);
 
     StoreKey deletedKey = null;
-    if (existing == null && putValueId != null) {
+    boolean storeKeyExists = existing != null && existing.content().action().exists();
+    if (!storeKeyExists && putValueId != null) {
       // Check for a Delete-op in the same commit, representing a rename operation.
       UUID expectedContentID = UUID.fromString(putValueId);
       deletedKey = deleted.remove(expectedContentID);
     }
-    if (existing != null && putValueId == null && deleted.containsValue(storeKey)) {
+    if (storeKeyExists && putValueId == null && deleted.containsValue(storeKey)) {
       // Check for a Delete-op with same key in the same commit, representing a re-add operation.
       deletedKey = storeKey;
     }

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

@@ -17,6 +17,7 @@
 
 import static com.google.common.collect.Streams.stream;
 import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.catchThrowable;
 import static org.assertj.core.api.Assertions.tuple;
@@ -34,7 +35,9 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.assertj.core.api.SoftAssertions;
 import org.assertj.core.api.junit.jupiter.InjectSoftAssertions;
@@ -907,6 +910,98 @@ void renames(OperationOrder order, boolean reuseContentKey, boolean reuseContent
     }
   }
 
+  @Test
+  void renameTwice() throws Exception {
+    BranchName branch = BranchName.of("main");
+
+    ContentKey key = ContentKey.of("table");
+    ContentKey keyBackup = ContentKey.of("table_backup");
+    ContentKey keyTemp = ContentKey.of("table_tmp");
+    List<ContentKey> keys = ImmutableList.of(key, keyTemp, keyBackup);
+
+    IcebergTable tableOld = IcebergTable.of("old", 1, 2, 3, 4);
+    IcebergTable tableNew = IcebergTable.of("new", 1, 2, 3, 4);
+
+    store()
+        .commit(
+            branch,
+            Optional.empty(),
+            CommitMeta.fromMessage("commit"),
+            singletonList(Put.of(key, tableOld)))
+        .getCommitHash();
+    tableOld = (IcebergTable) store().getValue(branch, key).content();
+
+    soft.assertThat(
+            store().getValues(branch, keys).entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().content())))
+        .hasSize(1)
+        .containsEntry(key, tableOld);
+
+    // create new "table_tmp"
+
+    store()
+        .commit(
+            branch,
+            Optional.empty(),
+            CommitMeta.fromMessage("new"),
+            singletonList(Put.of(keyTemp, tableNew)));
+    tableNew = (IcebergTable) store().getValue(branch, keyTemp).content();
+
+    soft.assertThat(
+            store().getValues(branch, keys).entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().content())))
+        .hasSize(2)
+        .containsEntry(key, tableOld)
+        .containsEntry(keyTemp, tableNew);
+
+    // rename "original" to "original_backup"
+
+    store()
+        .commit(
+            branch,
+            Optional.empty(),
+            CommitMeta.fromMessage("backup"),
+            ImmutableList.of(Delete.of(key), Put.of(keyBackup, tableOld)));
+
+    soft.assertThat(
+            store().getValues(branch, keys).entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().content())))
+        .hasSize(2)
+        .containsEntry(keyBackup, tableOld)
+        .containsEntry(keyTemp, tableNew);
+
+    // rename new "table_tmp" to "table"
+
+    store()
+        .commit(
+            branch,
+            Optional.empty(),
+            CommitMeta.fromMessage("rename new"),
+            ImmutableList.of(Delete.of(keyTemp), Put.of(key, tableNew)));
+
+    soft.assertThat(
+            store().getValues(branch, keys).entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().content())))
+        .hasSize(2)
+        .containsEntry(keyBackup, tableOld)
+        .containsEntry(key, tableNew);
+
+    // delete backup "table_backup"
+
+    store()
+        .commit(
+            branch,
+            Optional.empty(),
+            CommitMeta.fromMessage("delete"),
+            singletonList(Delete.of(keyBackup)));
+
+    soft.assertThat(
+            store().getValues(branch, keys).entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().content())))
+        .hasSize(1)
+        .containsEntry(key, tableNew);
+  }
+
   @Test
   void commitWithValidation() throws Exception {
     BranchName branch = BranchName.of("main");