added nosnapshotfound exception

Author: Blazer-007

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java

@@ -79,6 +79,12 @@ public TableNotFoundException(TableIdentifier tableId) {
     }
   }
 
+  public static class NoSnapshotFoundException extends IOException {
+    public NoSnapshotFoundException(TableIdentifier tableId) {
+      super("No Snapshot found: '" + tableId + "'");
+    }
+  }
+
   @Getter
   private final TableIdentifier tableId;
   /** allow the {@link IcebergCatalog} creating this table to qualify its {@link DatasetDescriptor#getName()} used for lineage, etc. */
@@ -97,19 +103,22 @@ public TableNotFoundException(TableIdentifier tableId) {
   /** @return metadata info limited to the most recent (current) snapshot */
   public IcebergSnapshotInfo getCurrentSnapshotInfo() throws IOException {
     TableMetadata current = accessTableMetadata();
-    return createSnapshotInfo(current.currentSnapshot(), Optional.of(current.metadataFileLocation()), Optional.of(current));
+    Snapshot currentSnapshot = accessCurrentSnapshot(current);
+    return createSnapshotInfo(currentSnapshot, Optional.of(current.metadataFileLocation()), Optional.of(current));
   }
 
   /** @return metadata info for most recent snapshot, wherein manifests and their child data files ARE NOT listed */
   public IcebergSnapshotInfo getCurrentSnapshotInfoOverviewOnly() throws IOException {
     TableMetadata current = accessTableMetadata();
-    return createSnapshotInfo(current.currentSnapshot(), Optional.of(current.metadataFileLocation()), Optional.of(current), true);
+    Snapshot currentSnapshot = accessCurrentSnapshot(current);
+    return createSnapshotInfo(currentSnapshot, Optional.of(current.metadataFileLocation()), Optional.of(current), true);
   }
 
   /** @return metadata info for all known snapshots, ordered historically, with *most recent last* */
   public Iterator<IcebergSnapshotInfo> getAllSnapshotInfosIterator() throws IOException {
     TableMetadata current = accessTableMetadata();
-    long currentSnapshotId = current.currentSnapshot().snapshotId();
+    Snapshot currentSnapshot = accessCurrentSnapshot(current);
+    long currentSnapshotId = currentSnapshot.snapshotId();
     List<Snapshot> snapshots = current.snapshots();
     return Iterators.transform(snapshots.iterator(), snapshot -> {
       try {
@@ -172,6 +181,12 @@ protected TableMetadata accessTableMetadata() throws TableNotFoundException {
     return Optional.ofNullable(current).orElseThrow(() -> new TableNotFoundException(this.tableId));
   }
 
+  /** @throws {@link IcebergTable.NoSnapshotFoundException} when table is empty i.e. table has zero snapshot */
+  protected Snapshot accessCurrentSnapshot(TableMetadata tableMetadata) throws NoSnapshotFoundException {
+    Snapshot currentSnapshot = tableMetadata.currentSnapshot();
+    return Optional.ofNullable(currentSnapshot).orElseThrow(() -> new NoSnapshotFoundException(this.tableId));
+  }
+
   protected IcebergSnapshotInfo createSnapshotInfo(Snapshot snapshot, Optional<String> metadataFileLocation, Optional<TableMetadata> currentTableMetadata)
       throws IOException {
     return createSnapshotInfo(snapshot, metadataFileLocation, currentTableMetadata, false);
@@ -239,8 +254,7 @@ protected void registerIcebergTable(TableMetadata srcMetadata, TableMetadata dst
   public List<DataFile> getPartitionSpecificDataFiles(Predicate<StructLike> icebergPartitionFilterPredicate)
       throws IOException {
     TableMetadata tableMetadata = accessTableMetadata();
-    Snapshot currentSnapshot = Optional.ofNullable(tableMetadata.currentSnapshot())
-        .orElseThrow(() -> new IOException(String.format("~%s~ No snapshots found", tableId)));
+    Snapshot currentSnapshot = accessCurrentSnapshot(tableMetadata);
     long currentSnapshotId = currentSnapshot.snapshotId();
     List<DataFile> knownDataFiles = new ArrayList<>();
     GrowthMilestoneTracker growthMilestoneTracker = new GrowthMilestoneTracker();
@@ -287,10 +301,10 @@ protected void overwritePartition(List<DataFile> dataFiles, String partitionColN
       return;
     }
     TableMetadata tableMetadata = accessTableMetadata();
-    Optional<Snapshot> currentSnapshot = Optional.ofNullable(tableMetadata.currentSnapshot());
-    if (currentSnapshot.isPresent()) {
-      log.info("~{}~ SnapshotId before overwrite: {}", tableId, currentSnapshot.get().snapshotId());
-    } else {
+    try {
+      Snapshot currentSnapshot = accessCurrentSnapshot(tableMetadata);
+      log.info("~{}~ SnapshotId before overwrite: {}", tableId, currentSnapshot.snapshotId());
+    } catch (NoSnapshotFoundException e) {
       log.warn("~{}~ No current snapshot found before overwrite", tableId);
     }
     OverwriteFiles overwriteFiles = this.table.newOverwrite();

gobblin-data-management/src/test/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTableTest.java

@@ -126,6 +126,14 @@ public void testGetCurrentSnapshotInfoOnBogusTable() throws IOException {
     Assert.fail("expected an exception when using table ID '" + bogusTableId + "'");
   }
 
+  /** Verify failure when attempting to get current snapshot info for an empty table */
+  @Test(expectedExceptions = IcebergTable.NoSnapshotFoundException.class)
+  public void testGetCurrentSnapshotInfoOnEmptyTable() throws IOException {
+    IcebergSnapshotInfo snapshotInfo = new IcebergTable(tableId, catalog.newTableOps(tableId), catalogUri,
+        catalog.loadTable(tableId)).getCurrentSnapshotInfo();
+    Assert.fail("expected an exception when using table ID '" + tableId + "'");
+  }
+
   /** Verify info about all (full) snapshots */
   @Test
   public void testGetAllSnapshotInfosIterator() throws IOException {
@@ -253,18 +261,6 @@ public void testGetPartitionSpecificDataFiles() throws IOException {
     Assert.assertEquals(icebergTable.getPartitionSpecificDataFiles(alwaysFalsePredicate).size(), 0);
   }
 
-  @Test
-  public void testGetPartitionSpecificDataFilesThrowsExceptionWhenNoSnapshotsFound() throws IOException {
-    IcebergTable icebergTable = new IcebergTable(tableId,
-        catalog.newTableOps(tableId),
-        catalogUri,
-        catalog.loadTable(tableId));
-    // Using AlwaysTrue Predicate to avoid mocking of predicate class
-    Predicate<StructLike> alwaysTruePredicate = partition -> true;
-    IOException exception = Assert.expectThrows(IOException.class, () -> icebergTable.getPartitionSpecificDataFiles(alwaysTruePredicate));
-    Assert.assertEquals(exception.getMessage(), String.format("~%s~ No snapshots found", tableId));
-  }
-
   /** Verify that overwritePartition replace data files belonging to given partition col and value */
   @Test
   public void testOverwritePartition() throws IOException {