REST: Don't reset snapshotLog when replacing table

core/src/main/java/org/apache/iceberg/MetadataUpdate.java

@@ -327,18 +327,24 @@ public void applyTo(TableMetadata.Builder metadataBuilder) {
 
   class RemoveSnapshotRef implements MetadataUpdate {
     private final String refName;
+    private final boolean purge;
 
-    public RemoveSnapshotRef(String refName) {
+    public RemoveSnapshotRef(String refName, boolean purge) {
       this.refName = refName;
+      this.purge = purge;
     }
 
     public String name() {
       return refName;
     }
 
+    public boolean purge() {
+      return purge;
+    }
+
     @Override
     public void applyTo(TableMetadata.Builder metadataBuilder) {
-      metadataBuilder.removeRef(refName);
+      metadataBuilder.removeRef(refName, purge);
     }
   }
 

core/src/main/java/org/apache/iceberg/MetadataUpdateParser.java

@@ -105,6 +105,9 @@ private MetadataUpdateParser() {}
   private static final String MAX_SNAPSHOT_AGE_MS = "max-snapshot-age-ms";
   private static final String MAX_REF_AGE_MS = "max-ref-age-ms";
 
+  // RemoveSnapshotRef
+  private static final String PURGE = "purge";
+
   // SetProperties
   // the REST API Spec defines "updates" but we initially used "updated",
   // thus we need to support reading both indefinitely
@@ -417,6 +420,7 @@ private static void writeSetSnapshotRef(MetadataUpdate.SetSnapshotRef update, Js
   private static void writeRemoveSnapshotRef(
       MetadataUpdate.RemoveSnapshotRef update, JsonGenerator gen) throws IOException {
     gen.writeStringField(REF_NAME, update.name());
+    gen.writeBooleanField(PURGE, update.purge());
   }
 
   private static void writeSetProperties(MetadataUpdate.SetProperties update, JsonGenerator gen)
@@ -548,7 +552,8 @@ private static MetadataUpdate readSetSnapshotRef(JsonNode node) {
 
   private static MetadataUpdate readRemoveSnapshotRef(JsonNode node) {
     String refName = JsonUtil.getString(REF_NAME, node);
-    return new MetadataUpdate.RemoveSnapshotRef(refName);
+    boolean purge = JsonUtil.getBool(PURGE, node);
+    return new MetadataUpdate.RemoveSnapshotRef(refName, purge);
   }
 
   private static MetadataUpdate readSetProperties(JsonNode node) {

core/src/main/java/org/apache/iceberg/RemoveSnapshots.java

@@ -201,7 +201,7 @@ private TableMetadata internalApply() {
 
     base.refs().keySet().stream()
         .filter(ref -> !retainedRefs.containsKey(ref))
-        .forEach(updatedMetaBuilder::removeRef);
+        .forEach(name -> updatedMetaBuilder.removeRef(name, true));
 
     base.snapshots().stream()
         .map(Snapshot::snapshotId)

core/src/main/java/org/apache/iceberg/TableMetadata.java

@@ -1283,25 +1283,27 @@ public Builder setRef(String name, SnapshotRef ref) {
       return this;
     }
 
-    public Builder removeRef(String name) {
+    public Builder removeRef(String name, boolean purge) {
       if (SnapshotRef.MAIN_BRANCH.equals(name)) {
         this.currentSnapshotId = -1;
-        snapshotLog.clear();
+        if (purge) {
+          snapshotLog.clear();
+        }
       }
 
       SnapshotRef ref = refs.remove(name);
       if (ref != null) {
-        changes.add(new MetadataUpdate.RemoveSnapshotRef(name));
+        changes.add(new MetadataUpdate.RemoveSnapshotRef(name, true));
       }
 
       return this;
     }
 
-    private Builder resetMainBranch() {
+    public Builder resetMainBranch() {
       this.currentSnapshotId = -1;
       SnapshotRef ref = refs.remove(SnapshotRef.MAIN_BRANCH);
       if (ref != null) {
-        changes.add(new MetadataUpdate.RemoveSnapshotRef(SnapshotRef.MAIN_BRANCH));
+        changes.add(new MetadataUpdate.RemoveSnapshotRef(SnapshotRef.MAIN_BRANCH, false));
       }
 
       return this;
@@ -1408,7 +1410,7 @@ private Builder rewriteSnapshotsInternal(Collection<Long> idsToRemove, boolean s
         }
       }
 
-      danglingRefs.forEach(this::removeRef);
+      danglingRefs.forEach(name -> removeRef(name, true));
 
       return this;
     }

core/src/main/java/org/apache/iceberg/UpdateSnapshotReferencesOperation.java

@@ -192,7 +192,7 @@ private TableMetadata internalApply() {
     Map<String, SnapshotRef> currRefs = base.refs();
     for (Map.Entry<String, SnapshotRef> currRefEntry : currRefs.entrySet()) {
       if (!updatedRefs.containsKey(currRefEntry.getKey())) {
-        updatedBuilder.removeRef(currRefEntry.getKey());
+        updatedBuilder.removeRef(currRefEntry.getKey(), true);
       }
     }
 

core/src/test/java/org/apache/iceberg/TestMetadataUpdateParser.java

@@ -434,16 +434,18 @@ public void testRemoveSnapshotsToJson() {
   public void testRemoveSnapshotRefFromJson() {
     String action = MetadataUpdateParser.REMOVE_SNAPSHOT_REF;
     String snapshotRef = "snapshot-ref";
-    String json = "{\"action\":\"remove-snapshot-ref\",\"ref-name\":\"snapshot-ref\"}";
-    MetadataUpdate expected = new MetadataUpdate.RemoveSnapshotRef(snapshotRef);
+    String json =
+        "{\"action\":\"remove-snapshot-ref\",\"ref-name\":\"snapshot-ref\",\"purge\":true}";
+    MetadataUpdate expected = new MetadataUpdate.RemoveSnapshotRef(snapshotRef, true);
     assertEquals(action, expected, MetadataUpdateParser.fromJson(json));
   }
 
   @Test
   public void testRemoveSnapshotRefToJson() {
     String snapshotRef = "snapshot-ref";
-    String expected = "{\"action\":\"remove-snapshot-ref\",\"ref-name\":\"snapshot-ref\"}";
-    MetadataUpdate actual = new MetadataUpdate.RemoveSnapshotRef(snapshotRef);
+    String expected =
+        "{\"action\":\"remove-snapshot-ref\",\"ref-name\":\"snapshot-ref\",\"purge\":false}";
+    MetadataUpdate actual = new MetadataUpdate.RemoveSnapshotRef(snapshotRef, false);
     assertThat(MetadataUpdateParser.toJson(actual))
         .as("RemoveSnapshotRef should convert to the correct JSON value")
         .isEqualTo(expected);

core/src/test/java/org/apache/iceberg/TestSnapshotLoading.java

@@ -194,7 +194,9 @@ public void testRemovedRefSnapshotFails() {
   @TestTemplate
   public void testBuildingNewMetadataTriggersSnapshotLoad() {
     TableMetadata newTableMetadata =
-        TableMetadata.buildFrom(latestTableMetadata).removeRef(SnapshotRef.MAIN_BRANCH).build();
+        TableMetadata.buildFrom(latestTableMetadata)
+            .removeRef(SnapshotRef.MAIN_BRANCH, true)
+            .build();
 
     verify(snapshotsSupplierMock, times(1)).get();
   }

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

@@ -38,6 +38,7 @@
 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;
@@ -2418,6 +2419,30 @@ 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();
+
+    assertThat(((BaseTable) table).operations().current().snapshotLog()).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);
+  }
+
   @Test
   public void testCleanupUncommitedFilesForCleanableFailures() {
     RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));