diff --git a/‎iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalCatalog.java‎
Lines changed: 5 additions & 5 deletions b/‎iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalCatalog.java‎
Lines changed: 5 additions & 5 deletions
diff --git a/‎iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalTableOperations.java‎
Lines changed: 40 additions & 141 deletions b/‎iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/OpenHouseInternalTableOperations.java‎
Lines changed: 40 additions & 141 deletions
diff --git a/‎iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/SnapshotInspector.java‎
Lines changed: 0 additions & 96 deletions b/‎iceberg/openhouse/internalcatalog/src/main/java/com/linkedin/openhouse/internal/catalog/SnapshotInspector.java‎
Lines changed: 0 additions & 96 deletions
@@ -59,21 +59,21 @@ public class OpenHouseInternalCatalog extends BaseMetastoreCatalog {
 
   @Autowired StorageType storageType;
 
-  @Autowired SnapshotInspector snapshotInspector;
-
   @Autowired HouseTableMapper houseTableMapper;
 
   @Autowired MeterRegistry meterRegistry;
 
   @Override
   protected TableOperations newTableOps(TableIdentifier tableIdentifier) {
+    FileIO fileIO = resolveFileIO(tableIdentifier);
+    MetricsReporter metricsReporter =
+        new MetricsReporter(this.meterRegistry, METRICS_PREFIX, Lists.newArrayList());
     return new OpenHouseInternalTableOperations(
         houseTableRepository,
-        resolveFileIO(tableIdentifier),
-        snapshotInspector,
+        fileIO,
         houseTableMapper,
         tableIdentifier,
-        new MetricsReporter(this.meterRegistry, METRICS_PREFIX, Lists.newArrayList()),
+        metricsReporter,
         fileIOManager);
   }
 
 
@@ -23,7 +23,6 @@
 import java.io.IOException;
 import java.time.Clock;
 import java.time.Instant;
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -34,8 +33,6 @@
 import java.util.stream.Collectors;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.collections.MapUtils;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.iceberg.BaseMetastoreTableOperations;
 import org.apache.iceberg.PartitionField;
@@ -44,7 +41,6 @@
 import org.apache.iceberg.SchemaParser;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.SnapshotRef;
-import org.apache.iceberg.SnapshotSummary;
 import org.apache.iceberg.SortDirection;
 import org.apache.iceberg.SortField;
 import org.apache.iceberg.SortOrder;
@@ -56,11 +52,11 @@
 import org.apache.iceberg.exceptions.BadRequestException;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.expressions.Term;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.relocated.com.google.common.base.Objects;
-import org.springframework.data.util.Pair;
 
 @AllArgsConstructor
 @Slf4j
@@ -70,8 +66,6 @@ public class OpenHouseInternalTableOperations extends BaseMetastoreTableOperatio
 
   FileIO fileIO;
 
-  SnapshotInspector snapshotInspector;
-
   HouseTableMapper houseTableMapper;
 
   TableIdentifier tableIdentifier;
@@ -267,32 +261,56 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
       String sortOrderJson = properties.remove(CatalogConstants.SORT_ORDER_KEY);
       logPropertiesMap(properties);
 
-      TableMetadata updatedMetadata = metadata.replaceProperties(properties);
+      TableMetadata metadataToCommit = metadata.replaceProperties(properties);
 
       if (sortOrderJson != null) {
-        SortOrder sortOrder = SortOrderParser.fromJson(updatedMetadata.schema(), sortOrderJson);
-        updatedMetadata = updatedMetadata.replaceSortOrder(sortOrder);
+        SortOrder sortOrder = SortOrderParser.fromJson(metadataToCommit.schema(), sortOrderJson);
+        metadataToCommit = metadataToCommit.replaceSortOrder(sortOrder);
       }
 
       if (serializedSnapshotsToPut != null) {
         List<Snapshot> snapshotsToPut =
             SnapshotsUtil.parseSnapshots(fileIO, serializedSnapshotsToPut);
-        Pair<List<Snapshot>, List<Snapshot>> snapshotsDiff =
-            SnapshotsUtil.symmetricDifferenceSplit(snapshotsToPut, updatedMetadata.snapshots());
-        List<Snapshot> appendedSnapshots = snapshotsDiff.getFirst();
-        List<Snapshot> deletedSnapshots = snapshotsDiff.getSecond();
-        snapshotInspector.validateSnapshotsUpdate(
-            updatedMetadata, appendedSnapshots, deletedSnapshots);
         Map<String, SnapshotRef> snapshotRefs =
             serializedSnapshotRefs == null
                 ? new HashMap<>()
                 : SnapshotsUtil.parseSnapshotRefs(serializedSnapshotRefs);
-        updatedMetadata =
-            maybeAppendSnapshots(updatedMetadata, appendedSnapshots, snapshotRefs, true);
-        updatedMetadata = maybeDeleteSnapshots(updatedMetadata, deletedSnapshots);
+
+        TableMetadata.Builder builder = TableMetadata.buildFrom(metadataToCommit);
+
+        // 1. Identify which snapshots are new vs existing
+        Set<Long> existingSnapshotIds =
+            metadataToCommit.snapshots().stream()
+                .map(Snapshot::snapshotId)
+                .collect(Collectors.toSet());
+        Set<Long> newSnapshotIds =
+            snapshotsToPut.stream().map(Snapshot::snapshotId).collect(Collectors.toSet());
+
+        // 2. Add new snapshots
+        snapshotsToPut.stream()
+            .filter(s -> !existingSnapshotIds.contains(s.snapshotId()))
+            .forEach(builder::addSnapshot);
+
+        // 3. Remove snapshots that are no longer present in the client payload
+        List<Long> toRemove =
+            existingSnapshotIds.stream()
+                .filter(id -> !newSnapshotIds.contains(id))
+                .collect(Collectors.toList());
+        if (!toRemove.isEmpty()) {
+          builder.removeSnapshots(toRemove);
+        }
+
+        // 4. Sync Refs: Remove refs not in payload, Set/Update refs from payload
+        metadataToCommit.refs().keySet().stream()
+            .filter(ref -> !snapshotRefs.containsKey(ref))
+            .forEach(builder::removeRef);
+
+        snapshotRefs.forEach(builder::setRef);
+
+        metadataToCommit = builder.build();
       }
 
-      final TableMetadata updatedMtDataRef = updatedMetadata;
+      final TableMetadata updatedMtDataRef = metadataToCommit;
       long metadataUpdateStartTime = System.currentTimeMillis();
       try {
         metricsReporter.executeWithStats(
@@ -314,7 +332,7 @@ updatedMtDataRef, io().newOutputFile(newMetadataLocation)),
         throw e;
       }
 
-      houseTable = houseTableMapper.toHouseTable(updatedMetadata, fileIO);
+      houseTable = houseTableMapper.toHouseTable(metadataToCommit, fileIO);
       if (base != null
           && (properties.containsKey(CatalogConstants.OPENHOUSE_TABLEID_KEY)
                   && !properties
@@ -358,7 +376,7 @@ updatedMtDataRef, io().newOutputFile(newMetadataLocation)),
             e);
       }
       throw new CommitFailedException(ioe);
-    } catch (InvalidIcebergSnapshotException e) {
+    } catch (InvalidIcebergSnapshotException | IllegalArgumentException | ValidationException e) {
       throw new BadRequestException(e, e.getMessage());
     } catch (CommitFailedException e) {
       throw e;
@@ -531,125 +549,6 @@ private void failIfRetryUpdate(Map<String, String> properties) {
     }
   }
 
-  public TableMetadata maybeDeleteSnapshots(
-      TableMetadata metadata, List<Snapshot> snapshotsToDelete) {
-    TableMetadata result = metadata;
-    if (CollectionUtils.isNotEmpty(snapshotsToDelete)) {
-      Set<Long> snapshotIds =
-          snapshotsToDelete.stream().map(Snapshot::snapshotId).collect(Collectors.toSet());
-      Map<String, String> updatedProperties = new HashMap<>(result.properties());
-      updatedProperties.put(
-          getCanonicalFieldName(CatalogConstants.DELETED_SNAPSHOTS),
-          snapshotsToDelete.stream()
-              .map(s -> Long.toString(s.snapshotId()))
-              .collect(Collectors.joining(",")));
-      result =
-          TableMetadata.buildFrom(result)
-              .setProperties(updatedProperties)
-              .build()
-              .removeSnapshotsIf(s -> snapshotIds.contains(s.snapshotId()));
-      metricsReporter.count(
-          InternalCatalogMetricsConstant.SNAPSHOTS_DELETED_CTR, snapshotsToDelete.size());
-    }
-    return result;
-  }
-
-  public TableMetadata maybeAppendSnapshots(
-      TableMetadata metadata,
-      List<Snapshot> snapshotsToAppend,
-      Map<String, SnapshotRef> snapshotRefs,
-      boolean recordAction) {
-    TableMetadata.Builder metadataBuilder = TableMetadata.buildFrom(metadata);
-    List<String> appendedSnapshots = new ArrayList<>();
-    List<String> stagedSnapshots = new ArrayList<>();
-    List<String> cherryPickedSnapshots = new ArrayList<>();
-    // Throw an exception if client sent request that included non-main branches in the
-    // snapshotRefs.
-    for (Map.Entry<String, SnapshotRef> entry : snapshotRefs.entrySet()) {
-      if (!entry.getKey().equals(SnapshotRef.MAIN_BRANCH)) {
-        throw new UnsupportedOperationException("OpenHouse supports only MAIN branch");
-      }
-    }
-    /**
-     * First check if there are new snapshots to be appended to current TableMetadata. If yes,
-     * following are the cases to be handled:
-     *
-     * <p>[1] A regular (non-wap) snapshot is being added to the MAIN branch.
-     *
-     * <p>[2] A staged (wap) snapshot is being created on top of current snapshot as its base.
-     * Recognized by STAGED_WAP_ID_PROP.
-     *
-     * <p>[3] A staged (wap) snapshot is being cherry picked to the MAIN branch wherein current
-     * snapshot in the MAIN branch is not the same as the base snapshot the staged (wap) snapshot
-     * was created on. Recognized by SOURCE_SNAPSHOT_ID_PROP. This case is called non-fast forward
-     * cherry pick.
-     *
-     * <p>In case no new snapshots are to be appended to current TableMetadata, there could be a
-     * cherrypick of a staged (wap) snapshot on top of the current snapshot in the MAIN branch which
-     * is the same as the base snapshot the staged (wap) snapshot was created on. This case is
-     * called fast forward cherry pick.
-     */
-    if (CollectionUtils.isNotEmpty(snapshotsToAppend)) {
-      for (Snapshot snapshot : snapshotsToAppend) {
-        snapshotInspector.validateSnapshot(snapshot);
-        if (snapshot.summary().containsKey(SnapshotSummary.STAGED_WAP_ID_PROP)) {
-          // a stage only snapshot using wap.id
-          metadataBuilder.addSnapshot(snapshot);
-          stagedSnapshots.add(String.valueOf(snapshot.snapshotId()));
-        } else if (snapshot.summary().containsKey(SnapshotSummary.SOURCE_SNAPSHOT_ID_PROP)) {
-          // a snapshot created on a non fast-forward cherry-pick snapshot
-          metadataBuilder.setBranchSnapshot(snapshot, SnapshotRef.MAIN_BRANCH);
-          appendedSnapshots.add(String.valueOf(snapshot.snapshotId()));
-          cherryPickedSnapshots.add(
-              String.valueOf(snapshot.summary().get(SnapshotSummary.SOURCE_SNAPSHOT_ID_PROP)));
-        } else {
-          // a regular snapshot
-          metadataBuilder.setBranchSnapshot(snapshot, SnapshotRef.MAIN_BRANCH);
-          appendedSnapshots.add(String.valueOf(snapshot.snapshotId()));
-        }
-      }
-    } else if (MapUtils.isNotEmpty(snapshotRefs)) {
-      // Updated ref in the main branch with no new snapshot means this is a
-      // fast-forward cherry-pick or rollback operation.
-      long newSnapshotId = snapshotRefs.get(SnapshotRef.MAIN_BRANCH).snapshotId();
-      // Either the current snapshot is null or the current snapshot is not equal
-      // to the new snapshot indicates an update. The first case happens when the
-      // stage/wap snapshot being cherry-picked is the first snapshot.
-      if (MapUtils.isEmpty(metadata.refs())
-          || metadata.refs().get(SnapshotRef.MAIN_BRANCH).snapshotId() != newSnapshotId) {
-        metadataBuilder.setBranchSnapshot(newSnapshotId, SnapshotRef.MAIN_BRANCH);
-        cherryPickedSnapshots.add(String.valueOf(newSnapshotId));
-      }
-    }
-    if (recordAction) {
-      Map<String, String> updatedProperties = new HashMap<>(metadata.properties());
-      if (CollectionUtils.isNotEmpty(appendedSnapshots)) {
-        updatedProperties.put(
-            getCanonicalFieldName(CatalogConstants.APPENDED_SNAPSHOTS),
-            appendedSnapshots.stream().collect(Collectors.joining(",")));
-        metricsReporter.count(
-            InternalCatalogMetricsConstant.SNAPSHOTS_ADDED_CTR, appendedSnapshots.size());
-      }
-      if (CollectionUtils.isNotEmpty(stagedSnapshots)) {
-        updatedProperties.put(
-            getCanonicalFieldName(CatalogConstants.STAGED_SNAPSHOTS),
-            stagedSnapshots.stream().collect(Collectors.joining(",")));
-        metricsReporter.count(
-            InternalCatalogMetricsConstant.SNAPSHOTS_STAGED_CTR, stagedSnapshots.size());
-      }
-      if (CollectionUtils.isNotEmpty(cherryPickedSnapshots)) {
-        updatedProperties.put(
-            getCanonicalFieldName(CatalogConstants.CHERRY_PICKED_SNAPSHOTS),
-            cherryPickedSnapshots.stream().collect(Collectors.joining(",")));
-        metricsReporter.count(
-            InternalCatalogMetricsConstant.SNAPSHOTS_CHERRY_PICKED_CTR,
-            cherryPickedSnapshots.size());
-      }
-      metadataBuilder.setProperties(updatedProperties);
-    }
-    return metadataBuilder.build();
-  }
-
   /** Helper function to dump contents for map in debugging mode. */
   private void logPropertiesMap(Map<String, String> map) {
     log.debug(" === Printing the table properties within doCommit method === ");