ice/watch: Recreate table if deleted (and --create-table/-p is used)

Author: shyiko

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

@@ -61,6 +61,7 @@
 import org.apache.iceberg.data.parquet.GenericParquetWriter;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.BadRequestException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.io.FileIO;
@@ -92,7 +93,7 @@ private Insert() {}
   // TODO: refactor
   public static void run(
       RESTCatalog catalog, TableIdentifier nsTable, String[] files, Options options)
-      throws IOException, InterruptedException {
+      throws NoSuchTableException, IOException, InterruptedException {
     if (files.length == 0) {
       // no work to be done
       return;

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

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.function.Supplier;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.rest.RESTCatalog;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -83,8 +84,6 @@ public static void run(
           // TODO: implement
         };
 
-    boolean createTableExecuted = false;
-
     //noinspection LoopConditionNotUpdatedInsideLoop
     do {
       List<Message> batch = new LinkedList<>();
@@ -117,22 +116,23 @@ public static void run(
           if (!insertBatch.isEmpty()) {
             logger.info("Inserting {}", insertBatch);
 
-            if (createTableIfNotExists && !createTableExecuted) {
-              if (!catalog.tableExists(nsTable)) {
-                CreateTable.run(
-                    catalog,
-                    nsTable,
-                    insertBatch.iterator().next(),
-                    null,
-                    true,
-                    options.s3NoSignRequest(),
-                    null,
-                    null);
+            try {
+              Insert.run(catalog, nsTable, insertBatch.toArray(String[]::new), options);
+            } catch (NoSuchTableException e) {
+              if (!createTableIfNotExists) {
+                throw e;
               }
-              createTableExecuted = true;
+              CreateTable.run(
+                catalog,
+                nsTable,
+                insertBatch.iterator().next(),
+                null,
+                true,
+                options.s3NoSignRequest(),
+                null,
+                null);
+              Insert.run(catalog, nsTable, insertBatch.toArray(String[]::new), options);
             }
-
-            Insert.run(catalog, nsTable, insertBatch.toArray(String[]::new), options);
           }
 
           confirmProcessed(sqs, sqsQueueURL, batch);