ice/watch: Tweak handling of deleted files

Author: shyiko

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

@@ -29,14 +29,17 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.BadRequestException;
+import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.mapping.MappingUtil;
 import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.mapping.NameMappingParser;
 import org.apache.iceberg.parquet.ParquetSchemaUtil;
 import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 import org.apache.parquet.schema.MessageType;
 import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
 import software.amazon.awssdk.utils.Lazy;
 
 public final class CreateTable {
@@ -69,7 +72,15 @@ public static void run(
     }
     try (var inputIO = Input.newIO(schemaFile, null, s3ClientLazy)) {
       InputFile inputFile = Input.newFile(schemaFile, catalog, inputIO);
-      MessageType type = Metadata.read(inputFile).getFileMetaData().getSchema();
+
+      ParquetMetadata metadata;
+      try {
+        metadata = Metadata.read(inputFile);
+      } catch (NoSuchKeyException e) { // S3FileInput
+        throw new NotFoundException(inputFile.location(), e);
+      } // rethrow NotFoundException
+
+      MessageType type = metadata.getFileMetaData().getSchema();
       Schema fileSchema = ParquetSchemaUtil.convert(type);
       try {
         Map<String, String> props = null;

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

@@ -83,6 +83,7 @@
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
 import software.amazon.awssdk.utils.Lazy;
 
 public final class Insert {
@@ -364,9 +365,14 @@ private static List<DataFile> processFile(
     InputFile inputFile = Input.newFile(file, catalog, inputIO == null ? tableIO : inputIO);
     ParquetMetadata metadata;
     try {
-      metadata = Metadata.read(inputFile);
+      try {
+        metadata = Metadata.read(inputFile);
+      } catch (NoSuchKeyException e) { // S3FileInput
+        throw new NotFoundException(inputFile.location(), e);
+      }
     } catch (NotFoundException e) {
       if (options.ignoreNotFound) {
+        logger.info("{}: not found (skipping)", file);
         return List.of();
       }
       throw e;

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

@@ -23,6 +23,7 @@
 import java.util.function.Supplier;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.rest.RESTCatalog;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -122,16 +123,27 @@ public static void run(
               if (!createTableIfNotExists) {
                 throw e;
               }
-              CreateTable.run(
-                catalog,
-                nsTable,
-                insertBatch.iterator().next(),
-                null,
-                true,
-                options.s3NoSignRequest(),
-                null,
-                null);
-              Insert.run(catalog, nsTable, insertBatch.toArray(String[]::new), options);
+              boolean retryInsert = true;
+              try {
+                CreateTable.run(
+                    catalog,
+                    nsTable,
+                    insertBatch.iterator().next(),
+                    null,
+                    true,
+                    options.s3NoSignRequest(),
+                    null,
+                    null);
+              } catch (NotFoundException nfe) {
+                if (!options.ignoreNotFound()) {
+                  throw nfe;
+                }
+                logger.info("Table not created ({} don't exist)", insertBatch);
+                retryInsert = false;
+              }
+              if (retryInsert) {
+                Insert.run(catalog, nsTable, insertBatch.toArray(String[]::new), options);
+              }
             }
           }