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/TableMetadata.java

@@ -1284,24 +1284,30 @@ public Builder setRef(String name, SnapshotRef ref) {
     }
 
     public Builder removeRef(String name) {
+      return removeRef(name, true);
+    }
+
+    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;

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/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));