ice/watch: Ignore deleted files

Author: shyiko

ice/src/main/java/com/altinity/ice/cli/Main.java

@@ -351,6 +351,7 @@ void insert(
               .s3NoSignRequest(s3NoSignRequest)
               .s3CopyObject(s3CopyObject)
               .assumeSorted(assumeSorted)
+              .ignoreNotFound(watchMode)
               .retryListFile(retryList)
               .partitionList(partitions)
               .sortOrderList(sortOrders)

ice/src/main/java/com/altinity/ice/cli/internal/cmd/Insert.java

@@ -62,6 +62,7 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.BadRequestException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.io.FileIO;
@@ -361,7 +362,15 @@ private static List<DataFile> processFile(
         };
 
     InputFile inputFile = Input.newFile(file, catalog, inputIO == null ? tableIO : inputIO);
-    ParquetMetadata metadata = Metadata.read(inputFile);
+    ParquetMetadata metadata;
+    try {
+      metadata = Metadata.read(inputFile);
+    } catch (NotFoundException e) {
+      if (options.ignoreNotFound) {
+        return List.of();
+      }
+      throw e;
+    }
 
     boolean sorted = options.assumeSorted;
     if (!sorted && sortOrder.isSorted()) {
@@ -724,6 +733,7 @@ public record Options(
       boolean s3NoSignRequest,
       boolean s3CopyObject,
       boolean assumeSorted,
+      boolean ignoreNotFound,
       @Nullable String retryListFile,
       @Nullable List<Main.IcePartition> partitionList,
       @Nullable List<Main.IceSortOrder> sortOrderList,
@@ -743,6 +753,7 @@ public static final class Builder {
       private boolean s3NoSignRequest;
       private boolean s3CopyObject;
       private boolean assumeSorted;
+      private boolean ignoreNotFound;
       String retryListFile;
       List<Main.IcePartition> partitionList = List.of();
       List<Main.IceSortOrder> sortOrderList = List.of();
@@ -795,6 +806,11 @@ public Builder assumeSorted(boolean assumeSorted) {
         return this;
       }
 
+      public Builder ignoreNotFound(boolean ignoreNotFound) {
+        this.ignoreNotFound = ignoreNotFound;
+        return this;
+      }
+
       public Builder retryListFile(String retryListFile) {
         this.retryListFile = retryListFile;
         return this;
@@ -826,6 +842,7 @@ public Options build() {
             s3NoSignRequest,
             s3CopyObject,
             assumeSorted,
+            ignoreNotFound,
             retryListFile,
             partitionList,
             sortOrderList,