Support for buffering inserts

examples/grafana/METRICS.md

@@ -120,6 +120,9 @@ These metrics track S3 event-driven file insertions.
 | `ice_watch_queue_receive_errors_total` | Counter | table, queue, queue_type | Total errors when receiving messages from queue |
 | `ice_watch_queue_delete_errors_total` | Counter | table, queue, queue_type | Total errors when deleting/acknowledging messages |
 | `ice_watch_message_parse_errors_total` | Counter | table, queue, queue_type | Total message parsing errors |
+| `ice_watch_buffer_files` | Gauge | table, queue, queue_type | Files accumulated and waiting to be committed |
+| `ice_watch_buffer_bytes` | Gauge | table, queue, queue_type | Total size of files accumulated and waiting to be committed |
+| `ice_watch_buffer_flushes_total` | Counter | table, queue, queue_type, trigger | Accumulated batches committed, by the threshold that triggered the commit |
 
 ### Maintenance Metrics
 

ice-rest-catalog/pom.xml

@@ -547,7 +547,7 @@
     <dependency>
       <groupId>com.github.shyiko.skedule</groupId>
       <artifactId>skedule</artifactId>
-      <version>0.4.0</version>
+      <version>${skedule.version}</version>
       <classifier>kalvanized</classifier>
     </dependency>
   </dependencies>

ice/pom.xml

@@ -526,6 +526,13 @@
       <artifactId>prometheus-metrics-exporter-servlet-jakarta</artifactId>
       <version>${prometheus.version}</version>
     </dependency>
+    <!-- skedule -->
+    <dependency>
+      <groupId>com.github.shyiko.skedule</groupId>
+      <artifactId>skedule</artifactId>
+      <version>${skedule.version}</version>
+      <classifier>kalvanized</classifier>
+    </dependency>
     <!-- picocli -->
     <dependency>
       <groupId>info.picocli</groupId>

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

@@ -26,6 +26,7 @@
 import com.altinity.ice.cli.internal.cmd.Files;
 import com.altinity.ice.cli.internal.cmd.Insert;
 import com.altinity.ice.cli.internal.cmd.InsertWatch;
+import com.altinity.ice.cli.internal.cmd.InsertWatchBuffer;
 import com.altinity.ice.cli.internal.cmd.ListNamespaces;
 import com.altinity.ice.cli.internal.cmd.ListPartitions;
 import com.altinity.ice.cli.internal.cmd.ListSnapshots;
@@ -557,7 +558,25 @@ void insert(
       @CommandLine.Option(
               names = {"--watch-debug-addr"},
               description = "")
-          String watchDebugAddr)
+          String watchDebugAddr,
+      @CommandLine.Option(
+              names = {"--watch-commit-schedule"},
+              description =
+                  "Accumulate incoming files and commit them as a single snapshot on this schedule,"
+                      + " in https://github.com/shyiko/skedule format, e.g. \"every 5 minutes\","
+                      + " \"every day 02:00\" (default: commit on every poll)")
+          String watchCommitSchedule,
+      @CommandLine.Option(
+              names = {"--watch-max-files"},
+              description = "Commit as soon as this many files are accumulated (default: no limit)",
+              defaultValue = "0")
+          int watchMaxFiles,
+      @CommandLine.Option(
+              names = {"--watch-max-bytes"},
+              description =
+                  "Commit as soon as accumulated files add up to this many bytes (default: no limit)",
+              defaultValue = "0")
+          long watchMaxBytes)
       throws IOException, InterruptedException {
     if (s3NoSignRequest && s3CopyObject) {
       throw new UnsupportedOperationException(
@@ -606,6 +625,11 @@ void insert(
       TableIdentifier tableId = TableIdentifier.parse(name);
       boolean watchMode = !Strings.isNullOrEmpty(watch);
 
+      if (!watchMode && (watchCommitSchedule != null || watchMaxFiles > 0 || watchMaxBytes > 0)) {
+        throw new IllegalArgumentException(
+            "--watch-commit-schedule/--watch-max-files/--watch-max-bytes require --watch");
+      }
+
       if (createTableIfNotExists && !watchMode) {
         CreateTable.run(
             catalog,
@@ -679,6 +703,7 @@ void insert(
             watchFireOnce,
             createTableIfNotExists,
             options,
+            new InsertWatchBuffer.BatchOptions(watchCommitSchedule, watchMaxFiles, watchMaxBytes),
             metricsEnabled);
       }
     }