Follow up to #77

Author: shyiko

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

@@ -271,12 +271,12 @@ void insert(
               arity = "0..*",
               paramLabel = "<files>",
               description = "/path/to/file.parquet")
-          String[] dataFiles,
+          String[] files,
       @CommandLine.Option(
-              names = "--from-file",
+              names = "--files-from",
               description =
                   "Read list of parquet files from the specified file (one file per line)")
-          String fromFile,
+          String filesFrom,
       @CommandLine.Option(
               names = "--no-copy",
               description = "Add files to catalog without copying them")
@@ -324,6 +324,11 @@ void insert(
                   "/path/to/file where to save list of files to retry"
                       + " (useful for retrying partially failed insert using `cat ice.retry | ice insert - --retry-list=ice.retry`)")
           String retryList,
+      @CommandLine.Option(
+              names = {"--retry-list-exit-code"},
+              description =
+                  "Exit code to return when insert produces non-empty --retry-list file (default: 0)")
+          int retryListExitCode,
       @CommandLine.Option(
               names = {"--partition"},
               description =
@@ -361,27 +366,30 @@ void insert(
       throw new UnsupportedOperationException(
           "--s3-no-sign-request + --s3-copy-object is not supported by AWS (see --help for details)");
     }
+    boolean filesSet = files != null && files.length > 0;
+    boolean filesFromSet = !Strings.isNullOrEmpty(filesFrom);
+    if (!filesSet && !filesFromSet) {
+      throw new IllegalArgumentException(
+          "At least one <files> argument or --files-from is required");
+    }
+    if (filesSet && filesFromSet) {
+      throw new IllegalArgumentException(
+          "<files> arguments and --files-from are mutually exclusive");
+    }
     setAWSRegion(s3Region);
     try (RESTCatalog catalog = loadCatalog()) {
-      // Handle --from-file option
-      if (!Strings.isNullOrEmpty(fromFile)) {
-        if (dataFiles != null && dataFiles.length > 0) {
-          throw new IllegalArgumentException("Cannot specify both --from-file and file arguments");
-        }
-        dataFiles = readInputFromFile(fromFile).toArray(new String[0]);
-        if (dataFiles.length == 0) {
+      if (filesFromSet) {
+        files = readInputFromFile(filesFrom).toArray(new String[0]);
+        if (files.length == 0) {
           logger.info("Nothing to insert (file empty)");
           return;
         }
-      } else if (dataFiles != null && dataFiles.length == 1 && "-".equals(dataFiles[0])) {
-        dataFiles = readInput().toArray(new String[0]);
-        if (dataFiles.length == 0) {
+      } else if (files.length == 1 && "-".equals(files[0])) {
+        files = readInput().toArray(new String[0]);
+        if (files.length == 0) {
           logger.info("Nothing to insert (stdin empty)");
           return;
         }
-      } else if (dataFiles == null || dataFiles.length == 0) {
-        throw new IllegalArgumentException(
-            "No files specified. Provide files as arguments or use --from-file");
       }
 
       List<IcePartition> partitions = null;
@@ -405,7 +413,7 @@ void insert(
         CreateTable.run(
             catalog,
             tableId,
-            dataFiles[0],
+            files[0],
             null,
             createTableIfNotExists,
             useVendedCredentials,
@@ -436,12 +444,14 @@ void insert(
               .build();
 
       if (!watchMode) {
-        Insert.Result result = Insert.run(catalog, tableId, dataFiles, options);
-        if (result.anyFilesFailed()) {
+        Insert.Result result = Insert.run(catalog, tableId, files, options);
+        if (!result.ok()) {
           logger.error(
-              "{} file(s) failed to insert. Check logs or retry list for details.",
-              result.failedCount());
-          System.exit(1);
+              "{}/{} file(s) failed to insert (see {})",
+              result.totalNumberOfFiles(),
+              result.numberOfFilesFailedToInsert(),
+              retryList);
+          System.exit(retryListExitCode);
         }
       } else {
         if (!Strings.isNullOrEmpty(watchDebugAddr)) {
@@ -459,7 +469,7 @@ void insert(
         }
 
         InsertWatch.run(
-            catalog, tableId, dataFiles, watch, watchFireOnce, createTableIfNotExists, options);
+            catalog, tableId, files, watch, watchFireOnce, createTableIfNotExists, options);
       }
     }
   }

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

@@ -35,7 +35,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -93,15 +92,20 @@ public final class Insert {
 
   private Insert() {}
 
-  public record Result(boolean anyFilesProcessed, boolean anyFilesFailed, int failedCount) {}
+  public record Result(int totalNumberOfFiles, int numberOfFilesFailedToInsert) {
+
+    public boolean ok() {
+      return numberOfFilesFailedToInsert == 0;
+    }
+  }
 
   // TODO: refactor
   public static Result run(
       RESTCatalog catalog, TableIdentifier nsTable, String[] files, Options options)
       throws NoSuchTableException, IOException, InterruptedException {
     if (files.length == 0) {
       // no work to be done
-      return new Result(false, false, 0);
+      return new Result(0, 0);
     }
 
     Table table = catalog.loadTable(nsTable);
@@ -169,10 +173,10 @@ public static Result run(
                     ? new RetryLog(options.retryListFile)
                     : null) {
           boolean atLeastOneFileAppended = false;
-          final AtomicInteger failedCount = new AtomicInteger(0);
 
           int numThreads = Math.min(options.threadCount(), filesExpanded.size());
           ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+          int failed = 0;
           try {
             var futures = new ArrayList<Future<List<DataFile>>>();
             for (final String file : filesExpanded) {
@@ -206,7 +210,6 @@ public static Result run(
                             logger.error(
                                 "{}: error (adding to retry list and continuing)", file, e);
                             retryLog.add(file);
-                            failedCount.incrementAndGet();
                             return Collections.emptyList();
                           } else {
                             throw new IOException(String.format("Error processing %s", file), e);
@@ -223,10 +226,10 @@ public static Result run(
                   appendOp.appendFile(df); // Only main thread appends now
                 }
               } catch (ExecutionException e) {
+                failed++;
                 if (retryLog == null) {
                   throw new IOException("Error processing file(s)", e.getCause());
                 }
-                failedCount.incrementAndGet();
               }
             }
           } finally {
@@ -252,8 +255,7 @@ public static Result run(
             logger.warn("Table commit skipped (--no-commit)");
           }
 
-          int finalFailedCount = failedCount.get();
-          return new Result(atLeastOneFileAppended, finalFailedCount > 0, finalFailedCount);
+          return new Result(filesExpanded.size(), failed);
         }
       } finally {
         if (s3ClientLazy.hasValue()) {

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

@@ -122,8 +122,11 @@ public static void run(
             try {
               Insert.Result result =
                   Insert.run(catalog, nsTable, insertBatch.toArray(String[]::new), options);
-              if (result.anyFilesFailed()) {
-                logger.warn("{} file(s) failed to insert in this batch", result.failedCount());
+              if (!result.ok()) {
+                logger.warn(
+                    "{}/{} file(s) failed to insert in this batch",
+                    result.totalNumberOfFiles(),
+                    result.numberOfFilesFailedToInsert());
               }
             } catch (NoSuchTableException e) {
               if (!createTableIfNotExists) {
@@ -151,8 +154,11 @@ public static void run(
               if (retryInsert) {
                 Insert.Result result =
                     Insert.run(catalog, nsTable, insertBatch.toArray(String[]::new), options);
-                if (result.anyFilesFailed()) {
-                  logger.warn("{} file(s) failed to insert in this batch", result.failedCount());
+                if (!result.ok()) {
+                  logger.warn(
+                      "{}/{} file(s) failed to insert in this batch",
+                      result.totalNumberOfFiles(),
+                      result.numberOfFilesFailedToInsert());
                 }
               }
             }