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();
  }
@Test
  void testAppendSnapshotsWithOldSnapshots() throws IOException {
    TableMetadata metadata =
        TableMetadata.buildFrom(BASE_TABLE_METADATA)
            .setPreviousFileLocation("tmp_location")
            .setLocation(BASE_TABLE_METADATA.metadataFileLocation())
            .build();
    // all snapshots are from the past and snapshots add should fail the validation
    List<Snapshot> snapshots = IcebergTestUtil.getSnapshots();
    Assertions.assertThrows(
        IllegalArgumentException.class,
        () ->
            openHouseInternalTableOperations.maybeAppendSnapshots(
                metadata, snapshots, ImmutableMap.of(), false));
    // the latest snapshots have larger timestamp than the previous metadata timestamp, so it should
    // pass the validation
    snapshots.addAll(IcebergTestUtil.getFutureSnapshots());
    openHouseInternalTableOperations.maybeAppendSnapshots(
        metadata, snapshots, ImmutableMap.of(), false);
  }