From 4e55282caabf11638bc8d6bb49eee68564e58228 Mon Sep 17 00:00:00 2001
From: Yuya Ebihara <ebyhry@gmail.com>
Date: Mon, 23 Dec 2024 10:57:08 +0900
Subject: [PATCH] Core: Don't reset snapshotLog in TableMetadata.removeRef

---
 .../org/apache/iceberg/TableMetadata.java     |  1 -
 .../org/apache/iceberg/TestTableMetadata.java | 18 +++++++++++
 .../apache/iceberg/rest/TestRESTCatalog.java  | 31 +++++++++++++++++++
 3 files changed, 49 insertions(+), 1 deletion(-)

diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 9f6ffbcc8714..19afb7af04aa 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -1286,7 +1286,6 @@ public Builder setRef(String name, SnapshotRef ref) {
     public Builder removeRef(String name) {
       if (SnapshotRef.MAIN_BRANCH.equals(name)) {
         this.currentSnapshotId = -1;
-        snapshotLog.clear();
       }
 
       SnapshotRef ref = refs.remove(name);
diff --git a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
index 6d066e8a654c..45aa211e5187 100644
--- a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
+++ b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
@@ -1693,6 +1693,24 @@ public void buildReplacementKeepsSnapshotLog() throws Exception {
         .containsExactlyElementsOf(metadata.snapshotLog());
   }
 
+  @Test
+  public void removeRefKeepsSnapshotLog() throws Exception {
+    TableMetadata metadata =
+        TableMetadataParser.fromJson(readTableMetadataInputFile("TableMetadataV2Valid.json"));
+    assertThat(metadata.currentSnapshot()).isNotNull();
+    assertThat(metadata.snapshots()).hasSize(2);
+    assertThat(metadata.snapshotLog()).hasSize(2);
+
+    TableMetadata removeRef =
+        TableMetadata.buildFrom(metadata).removeRef(SnapshotRef.MAIN_BRANCH).build();
+
+    assertThat(removeRef.currentSnapshot()).isNull();
+    assertThat(removeRef.snapshots()).hasSize(2).containsExactlyElementsOf(metadata.snapshots());
+    assertThat(removeRef.snapshotLog())
+        .hasSize(2)
+        .containsExactlyElementsOf(metadata.snapshotLog());
+  }
+
   @Test
   public void testConstructV3Metadata() {
     TableMetadata.newTableMetadata(
diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
index 768d6c3777ee..47f6efe92a61 100644
--- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
@@ -32,16 +32,19 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.BaseTransaction;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.HistoryEntry;
 import org.apache.iceberg.MetadataUpdate;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -2418,6 +2421,34 @@ public void testPaginationForListTables(int numberOfItems) {
             eq(ListTablesResponse.class));
   }
 
+  @Test
+  public void testReplaceTableKeepsSnapshotLog() {
+    RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
+    RESTCatalog catalog = catalog(adapter);
+
+    if (requiresNamespaceCreate()) {
+      catalog.createNamespace(TABLE.namespace());
+    }
+
+    catalog.createTable(TABLE, SCHEMA);
+
+    Table table = catalog.loadTable(TABLE);
+    table.newAppend().appendFile(FILE_A).commit();
+
+    List<HistoryEntry> snapshotLogBeforeReplace =
+        ((BaseTable) table).operations().current().snapshotLog();
+    assertThat(snapshotLogBeforeReplace).hasSize(1);
+
+    Transaction replaceTableTransaction = catalog.newReplaceTableTransaction(TABLE, SCHEMA, false);
+    replaceTableTransaction.newAppend().appendFile(FILE_A).commit();
+    replaceTableTransaction.commitTransaction();
+
+    table.refresh();
+    assertThat(((BaseTable) table).operations().current().snapshotLog())
+        .hasSize(2)
+        .containsAll(snapshotLogBeforeReplace);
+  }
+
   @Test
   public void testCleanupUncommitedFilesForCleanableFailures() {
     RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));