@@ -952,5 +952,19 @@ public abstract class TestExpireSnapshotsAction extends SparkTestBase {
 
     checkExpirationResults(1, 1, 1, result);
   }
+
+  @Test
+  public void testExpireOnEmptyTable() {
+
+    Set<String> deletedFiles = Sets.newHashSet();
+
+    // table has no data, testing ExpireSnapshots should not fail with no snapshot
+    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
+            .expireOlderThan(System.currentTimeMillis())
+            .deleteWith(deletedFiles::add)
+            .execute();
+
+    checkExpirationResults(0, 0, 0, result);
+  }
 }
 
/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.iceberg.actions;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.AssertHelpers;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.spark.SparkTestBase;
import org.apache.iceberg.types.Types;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

import static org.apache.iceberg.types.Types.NestedField.optional;

public abstract class TestExpireSnapshotsAction extends SparkTestBase {

  private static final HadoopTables TABLES = new HadoopTables(new Configuration());
  private static final Schema SCHEMA = new Schema(
      optional(1, "c1", Types.IntegerType.get()),
      optional(2, "c2", Types.StringType.get()),
      optional(3, "c3", Types.StringType.get())
  );

  private static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA).identity("c1").build();

  static final DataFile FILE_A = DataFiles.builder(SPEC)
      .withPath("/path/to/data-a.parquet")
      .withFileSizeInBytes(10)
      .withPartitionPath("c1=0") // easy way to set partition data for now
      .withRecordCount(1)
      .build();
  static final DataFile FILE_B = DataFiles.builder(SPEC)
      .withPath("/path/to/data-b.parquet")
      .withFileSizeInBytes(10)
      .withPartitionPath("c1=1") // easy way to set partition data for now
      .withRecordCount(1)
      .build();
  static final DataFile FILE_C = DataFiles.builder(SPEC)
      .withPath("/path/to/data-c.parquet")
      .withFileSizeInBytes(10)
      .withPartitionPath("c1=2") // easy way to set partition data for now
      .withRecordCount(1)
      .build();
  static final DataFile FILE_D = DataFiles.builder(SPEC)
      .withPath("/path/to/data-d.parquet")
      .withFileSizeInBytes(10)
      .withPartitionPath("c1=3") // easy way to set partition data for now
      .withRecordCount(1)
      .build();

  @Rule
  public TemporaryFolder temp = new TemporaryFolder();

  private File tableDir;
  private String tableLocation;
  private Table table;

  @Before
  public void setupTableLocation() throws Exception {
    this.tableDir = temp.newFolder();
    this.tableLocation = tableDir.toURI().toString();
    this.table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
    spark.conf().set("spark.sql.shuffle.partitions", 1);
  }

  private Long rightAfterSnapshot() {
    Long end = System.currentTimeMillis();
    while (end <= table.currentSnapshot().timestampMillis()) {
      end = System.currentTimeMillis();
    }
    return end;
  }

  private void checkExpirationResults(long expectedDatafiles, long expectedManifestsDeleted,
      long expectedManifestListsDeleted, ExpireSnapshotsActionResult results) {

    Assert.assertEquals("Incorrect number of manifest files deleted",
        (Long) expectedManifestsDeleted, results.manifestFilesDeleted());
    Assert.assertEquals("Incorrect number of datafiles deleted",
        (Long) expectedDatafiles, results.dataFilesDeleted());
    Assert.assertEquals("Incorrect number of manifest lists deleted",
        (Long) expectedManifestListsDeleted, results.manifestListsDeleted());
  }

  @Test
  public void testFilesCleaned() throws Exception {
    table.newFastAppend()
        .appendFile(FILE_A)
        .commit();

    table.newOverwrite()
        .deleteFile(FILE_A)
        .addFile(FILE_B)
        .commit();

    table.newFastAppend()
        .appendFile(FILE_C)
        .commit();

    long end = rightAfterSnapshot();

    ExpireSnapshotsActionResult results = Actions.forTable(table).expireSnapshots().expireOlderThan(end).execute();

    Assert.assertEquals("Table does not have 1 snapshot after expiration", 1, Iterables.size(table.snapshots()));

    checkExpirationResults(1L, 1L, 2L, results);
  }

  @Test
  public void dataFilesCleanupWithParallelTasks() throws IOException {

    table.newFastAppend()
        .appendFile(FILE_A)
        .commit();

    table.newFastAppend()
        .appendFile(FILE_B)
        .commit();

    table.newRewrite()
        .rewriteFiles(ImmutableSet.of(FILE_B), ImmutableSet.of(FILE_D))
        .commit();

    table.newRewrite()
        .rewriteFiles(ImmutableSet.of(FILE_A), ImmutableSet.of(FILE_C))
        .commit();

    long t4 = rightAfterSnapshot();

    Set<String> deletedFiles = Sets.newHashSet();
    Set<String> deleteThreads = ConcurrentHashMap.newKeySet();
    AtomicInteger deleteThreadsIndex = new AtomicInteger(0);

    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .executeDeleteWith(Executors.newFixedThreadPool(4, runnable -> {
          Thread thread = new Thread(runnable);
          thread.setName("remove-snapshot-" + deleteThreadsIndex.getAndIncrement());
          thread.setDaemon(true); // daemon threads will be terminated abruptly when the JVM exits
          return thread;
        }))
        .expireOlderThan(t4)
        .deleteWith(s -> {
          deleteThreads.add(Thread.currentThread().getName());
          deletedFiles.add(s);
        })
        .execute();

    // Verifies that the delete methods ran in the threads created by the provided ExecutorService ThreadFactory
    Assert.assertEquals(deleteThreads,
        Sets.newHashSet("remove-snapshot-0", "remove-snapshot-1", "remove-snapshot-2", "remove-snapshot-3"));

    Assert.assertTrue("FILE_A should be deleted", deletedFiles.contains(FILE_A.path().toString()));
    Assert.assertTrue("FILE_B should be deleted", deletedFiles.contains(FILE_B.path().toString()));

    checkExpirationResults(2L, 3L, 3L, result);
  }

  @Test
  public void testNoFilesDeletedWhenNoSnapshotsExpired() throws Exception {
    table.newFastAppend()
        .appendFile(FILE_A)
        .commit();

    ExpireSnapshotsActionResult results = Actions.forTable(table).expireSnapshots().execute();
    checkExpirationResults(0L, 0L, 0L, results);
  }

  @Test
  public void testCleanupRepeatedOverwrites() throws Exception {
    table.newFastAppend()
        .appendFile(FILE_A)
        .commit();

    for (int i = 0; i < 10; i++) {
      table.newOverwrite()
          .deleteFile(FILE_A)
          .addFile(FILE_B)
          .commit();

      table.newOverwrite()
          .deleteFile(FILE_B)
          .addFile(FILE_A)
          .commit();
    }

    long end = rightAfterSnapshot();
    ExpireSnapshotsActionResult results = Actions.forTable(table).expireSnapshots().expireOlderThan(end).execute();
    checkExpirationResults(1L, 39L, 20L, results);
  }

  @Test
  public void testRetainLastWithExpireOlderThan() {
    table.newAppend()
        .appendFile(FILE_A) // data_bucket=0
        .commit();
    long firstSnapshotId = table.currentSnapshot().snapshotId();
    long t1 = System.currentTimeMillis();
    while (t1 <= table.currentSnapshot().timestampMillis()) {
      t1 = System.currentTimeMillis();
    }

    table.newAppend()
        .appendFile(FILE_B) // data_bucket=1
        .commit();

    table.newAppend()
        .appendFile(FILE_C) // data_bucket=2
        .commit();

    long t3 = rightAfterSnapshot();

    // Retain last 2 snapshots
    Actions.forTable(table).expireSnapshots()
        .expireOlderThan(t3)
        .retainLast(2)
        .execute();

    Assert.assertEquals("Should have two snapshots.", 2, Lists.newArrayList(table.snapshots()).size());
    Assert.assertEquals("First snapshot should not present.", null, table.snapshot(firstSnapshotId));
  }

  @Test
  public void testExpireTwoSnapshotsById() throws Exception {
    table.newAppend()
        .appendFile(FILE_A) // data_bucket=0
        .commit();
    long firstSnapshotId = table.currentSnapshot().snapshotId();

    table.newAppend()
        .appendFile(FILE_B) // data_bucket=1
        .commit();

    long secondSnapshotID = table.currentSnapshot().snapshotId();

    table.newAppend()
        .appendFile(FILE_C) // data_bucket=2
        .commit();

    // Retain last 2 snapshots
    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireSnapshotId(firstSnapshotId)
        .expireSnapshotId(secondSnapshotID)
        .execute();

    Assert.assertEquals("Should have one snapshots.", 1, Lists.newArrayList(table.snapshots()).size());
    Assert.assertEquals("First snapshot should not present.", null, table.snapshot(firstSnapshotId));
    Assert.assertEquals("Second snapshot should not be present.", null, table.snapshot(secondSnapshotID));

    checkExpirationResults(0L, 0L, 2L, result);
  }

  @Test
  public void testRetainLastWithExpireById() {
    table.newAppend()
        .appendFile(FILE_A) // data_bucket=0
        .commit();
    long firstSnapshotId = table.currentSnapshot().snapshotId();

    table.newAppend()
        .appendFile(FILE_B) // data_bucket=1
        .commit();

    table.newAppend()
        .appendFile(FILE_C) // data_bucket=2
        .commit();

    // Retain last 3 snapshots, but explicitly remove the first snapshot
    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireSnapshotId(firstSnapshotId)
        .retainLast(3)
        .execute();

    Assert.assertEquals("Should have two snapshots.", 2, Lists.newArrayList(table.snapshots()).size());
    Assert.assertEquals("First snapshot should not present.", null, table.snapshot(firstSnapshotId));
    checkExpirationResults(0L, 0L, 1L, result);
  }

  @Test
  public void testRetainLastWithTooFewSnapshots() {
    table.newAppend()
        .appendFile(FILE_A) // data_bucket=0
        .appendFile(FILE_B) // data_bucket=1
        .commit();
    long firstSnapshotId = table.currentSnapshot().snapshotId();

    table.newAppend()
        .appendFile(FILE_C) // data_bucket=2
        .commit();

    long t2 = rightAfterSnapshot();

    // Retain last 3 snapshots
    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(t2)
        .retainLast(3)
        .execute();

    Assert.assertEquals("Should have two snapshots", 2, Lists.newArrayList(table.snapshots()).size());
    Assert.assertEquals("First snapshot should still present",
        firstSnapshotId, table.snapshot(firstSnapshotId).snapshotId());
    checkExpirationResults(0L, 0L, 0L, result);
  }

  @Test
  public void testRetainLastKeepsExpiringSnapshot() {
    table.newAppend()
        .appendFile(FILE_A) // data_bucket=0
        .commit();

    table.newAppend()
        .appendFile(FILE_B) // data_bucket=1
        .commit();

    Snapshot secondSnapshot = table.currentSnapshot();

    table.newAppend()
        .appendFile(FILE_C) // data_bucket=2
        .commit();

    table.newAppend()
        .appendFile(FILE_D) // data_bucket=3
        .commit();

    // Retain last 2 snapshots and expire older than t3
    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(secondSnapshot.timestampMillis())
        .retainLast(2)
        .execute();

    Assert.assertEquals("Should have three snapshots.", 3, Lists.newArrayList(table.snapshots()).size());
    Assert.assertNotNull("Second snapshot should present.", table.snapshot(secondSnapshot.snapshotId()));
    checkExpirationResults(0L, 0L, 1L, result);
  }

  @Test
  public void testExpireOlderThanMultipleCalls() {
    table.newAppend()
        .appendFile(FILE_A) // data_bucket=0
        .commit();

    table.newAppend()
        .appendFile(FILE_B) // data_bucket=1
        .commit();

    Snapshot secondSnapshot = table.currentSnapshot();

    table.newAppend()
        .appendFile(FILE_C) // data_bucket=2
        .commit();

    Snapshot thirdSnapshot = table.currentSnapshot();

    // Retain last 2 snapshots and expire older than t3
    ExpireSnapshotsActionResult result =  Actions.forTable(table).expireSnapshots()
        .expireOlderThan(secondSnapshot.timestampMillis())
        .expireOlderThan(thirdSnapshot.timestampMillis())
        .execute();

    Assert.assertEquals("Should have one snapshots.", 1, Lists.newArrayList(table.snapshots()).size());
    Assert.assertNull("Second snapshot should not present.", table.snapshot(secondSnapshot.snapshotId()));
    checkExpirationResults(0L, 0L, 2L, result);
  }

  @Test
  public void testRetainLastMultipleCalls() {
    table.newAppend()
        .appendFile(FILE_A) // data_bucket=0
        .commit();

    table.newAppend()
        .appendFile(FILE_B) // data_bucket=1
        .commit();

    Snapshot secondSnapshot = table.currentSnapshot();

    table.newAppend()
        .appendFile(FILE_C) // data_bucket=2
        .commit();

    long t3 = rightAfterSnapshot();

    // Retain last 2 snapshots and expire older than t3
    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(t3)
        .retainLast(2)
        .retainLast(1)
        .execute();

    Assert.assertEquals("Should have one snapshots.", 1, Lists.newArrayList(table.snapshots()).size());
    Assert.assertNull("Second snapshot should not present.", table.snapshot(secondSnapshot.snapshotId()));
    checkExpirationResults(0L, 0L, 2L, result);
  }

  @Test
  public void testRetainZeroSnapshots() {
    AssertHelpers.assertThrows("Should fail retain 0 snapshots " +
            "because number of snapshots to retain cannot be zero",
        IllegalArgumentException.class,
        "Number of snapshots to retain must be at least 1, cannot be: 0",
        () -> Actions.forTable(table).expireSnapshots().retainLast(0).execute());
  }

  @Test
  public void testScanExpiredManifestInValidSnapshotAppend() {
    table.newAppend()
        .appendFile(FILE_A)
        .appendFile(FILE_B)
        .commit();

    table.newOverwrite()
        .addFile(FILE_C)
        .deleteFile(FILE_A)
        .commit();

    table.newAppend()
        .appendFile(FILE_D)
        .commit();

    long t3 = rightAfterSnapshot();

    Set<String> deletedFiles = Sets.newHashSet();

    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(t3)
        .deleteWith(deletedFiles::add)
        .execute();

    Assert.assertTrue("FILE_A should be deleted", deletedFiles.contains(FILE_A.path().toString()));
    checkExpirationResults(1L, 1L, 2L, result);
  }

  @Test
  public void testScanExpiredManifestInValidSnapshotFastAppend() {
    table.updateProperties()
        .set(TableProperties.MANIFEST_MERGE_ENABLED, "true")
        .set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "1")
        .commit();

    table.newAppend()
        .appendFile(FILE_A)
        .appendFile(FILE_B)
        .commit();

    table.newOverwrite()
        .addFile(FILE_C)
        .deleteFile(FILE_A)
        .commit();

    table.newFastAppend()
        .appendFile(FILE_D)
        .commit();

    long t3 = rightAfterSnapshot();

    Set<String> deletedFiles = Sets.newHashSet();

    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(t3)
        .deleteWith(deletedFiles::add)
        .execute();

    Assert.assertTrue("FILE_A should be deleted", deletedFiles.contains(FILE_A.path().toString()));
    checkExpirationResults(1L, 1L, 2L, result);
  }

  /**
   * Test on table below, and expiring the staged commit `B` using `expireOlderThan` API.
   * Table: A - C
   *          ` B (staged)
   */
  @Test
  public void testWithExpiringDanglingStageCommit() {
    // `A` commit
    table.newAppend()
        .appendFile(FILE_A)
        .commit();

    // `B` staged commit
    table.newAppend()
        .appendFile(FILE_B)
        .stageOnly()
        .commit();

    TableMetadata base = ((BaseTable) table).operations().current();
    Snapshot snapshotA = base.snapshots().get(0);
    Snapshot snapshotB = base.snapshots().get(1);

    // `C` commit
    table.newAppend()
        .appendFile(FILE_C)
        .commit();

    Set<String> deletedFiles = new HashSet<>();

    // Expire all commits including dangling staged snapshot.
    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .deleteWith(deletedFiles::add)
        .expireOlderThan(snapshotB.timestampMillis() + 1)
        .execute();

    checkExpirationResults(1L, 1L, 2L, result);

    Set<String> expectedDeletes = new HashSet<>();
    expectedDeletes.add(snapshotA.manifestListLocation());

    // Files should be deleted of dangling staged snapshot
    snapshotB.addedFiles().forEach(i -> {
      expectedDeletes.add(i.path().toString());
    });

    // ManifestList should be deleted too
    expectedDeletes.add(snapshotB.manifestListLocation());
    snapshotB.dataManifests().forEach(file -> {
      // Only the manifest of B should be deleted.
      if (file.snapshotId() == snapshotB.snapshotId()) {
        expectedDeletes.add(file.path());
      }
    });
    Assert.assertSame("Files deleted count should be expected", expectedDeletes.size(), deletedFiles.size());
    // Take the diff
    expectedDeletes.removeAll(deletedFiles);
    Assert.assertTrue("Exactly same files should be deleted", expectedDeletes.isEmpty());
  }

  /**
   * Expire cherry-pick the commit as shown below, when `B` is in table's current state
   *  Table:
   *  A - B - C <--current snapshot
   *   `- D (source=B)
   */
  @Test
  public void testWithCherryPickTableSnapshot() {
    // `A` commit
    table.newAppend()
        .appendFile(FILE_A)
        .commit();
    Snapshot snapshotA = table.currentSnapshot();

    // `B` commit
    Set<String> deletedAFiles = new HashSet<>();
    table.newOverwrite()
        .addFile(FILE_B)
        .deleteFile(FILE_A)
        .deleteWith(deletedAFiles::add)
        .commit();
    Assert.assertTrue("No files should be physically deleted", deletedAFiles.isEmpty());

    // pick the snapshot 'B`
    Snapshot snapshotB = table.currentSnapshot();

    // `C` commit to let cherry-pick take effect, and avoid fast-forward of `B` with cherry-pick
    table.newAppend()
        .appendFile(FILE_C)
        .commit();
    Snapshot snapshotC = table.currentSnapshot();

    // Move the table back to `A`
    table.manageSnapshots()
        .setCurrentSnapshot(snapshotA.snapshotId())
        .commit();

    // Generate A -> `D (B)`
    table.manageSnapshots()
        .cherrypick(snapshotB.snapshotId())
        .commit();
    Snapshot snapshotD = table.currentSnapshot();

    // Move the table back to `C`
    table.manageSnapshots()
        .setCurrentSnapshot(snapshotC.snapshotId())
        .commit();
    List<String> deletedFiles = new ArrayList<>();

    // Expire `C`
    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .deleteWith(deletedFiles::add)
        .expireOlderThan(snapshotC.timestampMillis() + 1)
        .execute();

    // Make sure no dataFiles are deleted for the B, C, D snapshot
    Lists.newArrayList(snapshotB, snapshotC, snapshotD).forEach(i -> {
      i.addedFiles().forEach(item -> {
        Assert.assertFalse(deletedFiles.contains(item.path().toString()));
      });
    });

    checkExpirationResults(1L, 2L, 2L, result);
  }

  /**
   * Test on table below, and expiring `B` which is not in current table state.
   *  1) Expire `B`
   *  2) All commit
   * Table: A - C - D (B)
   *          ` B (staged)
   */
  @Test
  public void testWithExpiringStagedThenCherrypick() {
    // `A` commit
    table.newAppend()
        .appendFile(FILE_A)
        .commit();

    // `B` commit
    table.newAppend()
        .appendFile(FILE_B)
        .stageOnly()
        .commit();

    // pick the snapshot that's staged but not committed
    TableMetadata base = ((BaseTable) table).operations().current();
    Snapshot snapshotB = base.snapshots().get(1);

    // `C` commit to let cherry-pick take effect, and avoid fast-forward of `B` with cherry-pick
    table.newAppend()
        .appendFile(FILE_C)
        .commit();

    // `D (B)` cherry-pick commit
    table.manageSnapshots()
        .cherrypick(snapshotB.snapshotId())
        .commit();

    base = ((BaseTable) table).operations().current();
    Snapshot snapshotD = base.snapshots().get(3);

    List<String> deletedFiles = new ArrayList<>();

    // Expire `B` commit.
    ExpireSnapshotsActionResult firstResult = Actions.forTable(table).expireSnapshots()
        .deleteWith(deletedFiles::add)
        .expireSnapshotId(snapshotB.snapshotId())
        .execute();

    // Make sure no dataFiles are deleted for the staged snapshot
    Lists.newArrayList(snapshotB).forEach(i -> {
      i.addedFiles().forEach(item -> {
        Assert.assertFalse(deletedFiles.contains(item.path().toString()));
      });
    });
    checkExpirationResults(0L, 1L, 1L, firstResult);

    // Expire all snapshots including cherry-pick
    ExpireSnapshotsActionResult secondResult = Actions.forTable(table).expireSnapshots()
        .deleteWith(deletedFiles::add)
        .expireOlderThan(table.currentSnapshot().timestampMillis() + 1)
        .execute();

    // Make sure no dataFiles are deleted for the staged and cherry-pick
    Lists.newArrayList(snapshotB, snapshotD).forEach(i -> {
      i.addedFiles().forEach(item -> {
        Assert.assertFalse(deletedFiles.contains(item.path().toString()));
      });
    });
    checkExpirationResults(0L, 0L, 2L, secondResult);
  }


  @Test
  public void testExpireOlderThan() {
    table.newAppend()
        .appendFile(FILE_A)
        .commit();

    Snapshot firstSnapshot = table.currentSnapshot();

    rightAfterSnapshot();

    table.newAppend()
        .appendFile(FILE_B)
        .commit();

    long snapshotId = table.currentSnapshot().snapshotId();

    long tAfterCommits = rightAfterSnapshot();

    Set<String> deletedFiles = Sets.newHashSet();

    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(tAfterCommits)
        .deleteWith(deletedFiles::add)
        .execute();

    Assert.assertEquals("Expire should not change current snapshot", snapshotId, table.currentSnapshot().snapshotId());
    Assert.assertNull("Expire should remove the oldest snapshot", table.snapshot(firstSnapshot.snapshotId()));
    Assert.assertEquals("Should remove only the expired manifest list location",
        Sets.newHashSet(firstSnapshot.manifestListLocation()), deletedFiles);

    checkExpirationResults(0, 0, 1, result);
  }

  @Test
  public void testExpireOlderThanWithDelete() {
    table.newAppend()
        .appendFile(FILE_A)
        .commit();

    Snapshot firstSnapshot = table.currentSnapshot();
    Assert.assertEquals("Should create one manifest",
        1, firstSnapshot.allManifests().size());

    rightAfterSnapshot();

    table.newDelete()
        .deleteFile(FILE_A)
        .commit();

    Snapshot secondSnapshot = table.currentSnapshot();
    Assert.assertEquals("Should create replace manifest with a rewritten manifest",
        1, secondSnapshot.allManifests().size());

    table.newAppend()
        .appendFile(FILE_B)
        .commit();

    rightAfterSnapshot();

    long snapshotId = table.currentSnapshot().snapshotId();

    long tAfterCommits = rightAfterSnapshot();

    Set<String> deletedFiles = Sets.newHashSet();

    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(tAfterCommits)
        .deleteWith(deletedFiles::add)
        .execute();

    Assert.assertEquals("Expire should not change current snapshot", snapshotId, table.currentSnapshot().snapshotId());
    Assert.assertNull("Expire should remove the oldest snapshot", table.snapshot(firstSnapshot.snapshotId()));
    Assert.assertNull("Expire should remove the second oldest snapshot", table.snapshot(secondSnapshot.snapshotId()));

    Assert.assertEquals("Should remove expired manifest lists and deleted data file",
        Sets.newHashSet(
            firstSnapshot.manifestListLocation(), // snapshot expired
            firstSnapshot.allManifests().get(0).path(), // manifest was rewritten for delete
            secondSnapshot.manifestListLocation(), // snapshot expired
            secondSnapshot.allManifests().get(0).path(), // manifest contained only deletes, was dropped
            FILE_A.path()), // deleted
        deletedFiles);

    checkExpirationResults(1, 2, 2, result);
  }

  @Test
  public void testExpireOlderThanWithDeleteInMergedManifests() {
    // merge every commit
    table.updateProperties()
        .set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "0")
        .commit();

    table.newAppend()
        .appendFile(FILE_A)
        .appendFile(FILE_B)
        .commit();

    Snapshot firstSnapshot = table.currentSnapshot();
    Assert.assertEquals("Should create one manifest",
        1, firstSnapshot.allManifests().size());

    rightAfterSnapshot();

    table.newDelete()
        .deleteFile(FILE_A) // FILE_B is still in the dataset
        .commit();

    Snapshot secondSnapshot = table.currentSnapshot();
    Assert.assertEquals("Should replace manifest with a rewritten manifest",
        1, secondSnapshot.allManifests().size());

    table.newFastAppend() // do not merge to keep the last snapshot's manifest valid
        .appendFile(FILE_C)
        .commit();

    rightAfterSnapshot();

    long snapshotId = table.currentSnapshot().snapshotId();

    long tAfterCommits = rightAfterSnapshot();

    Set<String> deletedFiles = Sets.newHashSet();

    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(tAfterCommits)
        .deleteWith(deletedFiles::add)
        .execute();

    Assert.assertEquals("Expire should not change current snapshot", snapshotId, table.currentSnapshot().snapshotId());
    Assert.assertNull("Expire should remove the oldest snapshot", table.snapshot(firstSnapshot.snapshotId()));
    Assert.assertNull("Expire should remove the second oldest snapshot", table.snapshot(secondSnapshot.snapshotId()));

    Assert.assertEquals("Should remove expired manifest lists and deleted data file",
        Sets.newHashSet(
            firstSnapshot.manifestListLocation(), // snapshot expired
            firstSnapshot.allManifests().get(0).path(), // manifest was rewritten for delete
            secondSnapshot.manifestListLocation(), // snapshot expired
            FILE_A.path()), // deleted
        deletedFiles);

    checkExpirationResults(1, 1, 2, result);
  }

  @Test
  public void testExpireOlderThanWithRollback() {
    // merge every commit
    table.updateProperties()
        .set(TableProperties.MANIFEST_MIN_MERGE_COUNT, "0")
        .commit();

    table.newAppend()
        .appendFile(FILE_A)
        .appendFile(FILE_B)
        .commit();

    Snapshot firstSnapshot = table.currentSnapshot();
    Assert.assertEquals("Should create one manifest",
        1, firstSnapshot.allManifests().size());

    rightAfterSnapshot();

    table.newDelete()
        .deleteFile(FILE_B)
        .commit();

    Snapshot secondSnapshot = table.currentSnapshot();
    Set<ManifestFile> secondSnapshotManifests = Sets.newHashSet(secondSnapshot.allManifests());
    secondSnapshotManifests.removeAll(firstSnapshot.allManifests());
    Assert.assertEquals("Should add one new manifest for append", 1, secondSnapshotManifests.size());

    table.manageSnapshots()
        .rollbackTo(firstSnapshot.snapshotId())
        .commit();

    rightAfterSnapshot();

    long snapshotId = table.currentSnapshot().snapshotId();

    long tAfterCommits = rightAfterSnapshot();

    Set<String> deletedFiles = Sets.newHashSet();

    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(tAfterCommits)
        .deleteWith(deletedFiles::add)
        .execute();

    Assert.assertEquals("Expire should not change current snapshot", snapshotId, table.currentSnapshot().snapshotId());
    Assert.assertNotNull("Expire should keep the oldest snapshot, current", table.snapshot(firstSnapshot.snapshotId()));
    Assert.assertNull("Expire should remove the orphaned snapshot", table.snapshot(secondSnapshot.snapshotId()));

    Assert.assertEquals("Should remove expired manifest lists and reverted appended data file",
        Sets.newHashSet(
            secondSnapshot.manifestListLocation(), // snapshot expired
            Iterables.getOnlyElement(secondSnapshotManifests).path()), // manifest is no longer referenced
        deletedFiles);

    checkExpirationResults(0, 1, 1, result);
  }

  @Test
  public void testExpireOlderThanWithRollbackAndMergedManifests() {
    table.newAppend()
        .appendFile(FILE_A)
        .commit();

    Snapshot firstSnapshot = table.currentSnapshot();
    Assert.assertEquals("Should create one manifest",
        1, firstSnapshot.allManifests().size());

    rightAfterSnapshot();

    table.newAppend()
        .appendFile(FILE_B)
        .commit();

    Snapshot secondSnapshot = table.currentSnapshot();
    Set<ManifestFile> secondSnapshotManifests = Sets.newHashSet(secondSnapshot.allManifests());
    secondSnapshotManifests.removeAll(firstSnapshot.allManifests());
    Assert.assertEquals("Should add one new manifest for append", 1, secondSnapshotManifests.size());

    table.manageSnapshots()
        .rollbackTo(firstSnapshot.snapshotId())
        .commit();

    rightAfterSnapshot();

    long snapshotId = table.currentSnapshot().snapshotId();

    long tAfterCommits = rightAfterSnapshot();

    Set<String> deletedFiles = Sets.newHashSet();

    ExpireSnapshotsActionResult result = Actions.forTable(table).expireSnapshots()
        .expireOlderThan(tAfterCommits)
        .deleteWith(deletedFiles::add)
        .execute();

    Assert.assertEquals("Expire should not change current snapshot", snapshotId, table.currentSnapshot().snapshotId());
    Assert.assertNotNull("Expire should keep the oldest snapshot, current", table.snapshot(firstSnapshot.snapshotId()));
    Assert.assertNull("Expire should remove the orphaned snapshot", table.snapshot(secondSnapshot.snapshotId()));

    Assert.assertEquals("Should remove expired manifest lists and reverted appended data file",
        Sets.newHashSet(
            secondSnapshot.manifestListLocation(), // snapshot expired
            Iterables.getOnlyElement(secondSnapshotManifests).path(), // manifest is no longer referenced
            FILE_B.path()), // added, but rolled back
        deletedFiles);

    checkExpirationResults(1, 1, 1, result);
  }
}


nit: let's remove this extra empty line at the beginning