feat(loader): update loading logic & enhance failure case

.github/workflows/auto-pr-review.yml

@@ -0,0 +1,35 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: "Auto PR Commenter"
+
+on:
+  pull_request_target:
+    types: [opened]
+
+jobs:
+  add-review-comment:
+    runs-on: ubuntu-latest
+    permissions:
+      pull-requests: write
+    steps:
+      - name: Add review comment
+        uses: peter-evans/create-or-update-comment@v4
+        with:
+          issue-number: ${{ github.event.pull_request.number }}
+          body: |
+            @codecov-ai-reviewer review

hugegraph-loader/src/main/java/org/apache/hugegraph/loader/HugeGraphLoader.java

@@ -662,24 +662,21 @@ private List<InputTaskItem> prepareTaskItems(List<InputStruct> structs,
     }
 
     private void loadStructs(List<InputStruct> structs) {
-        int parallelCount = this.context.options().parallelCount;
+        int parallelThreads = this.context.options().parallelThreads;
         if (structs.size() == 0) {
             return;
         }
-        if (parallelCount <= 0) {
-            parallelCount = Math.min(structs.size(), Runtime.getRuntime().availableProcessors() * 2);
-        }
 
         boolean scatter = this.context.options().scatterSources;
 
         LOG.info("{} threads for loading {} structs, from {} to {} in {} mode",
-                 parallelCount, structs.size(), this.context.options().startFile,
+                 parallelThreads, structs.size(), this.context.options().startFile,
                  this.context.options().endFile,
                  scatter ? "scatter" : "sequential");
 
         ExecutorService loadService = null;
         try {
-            loadService = ExecutorUtil.newFixedThreadPool(parallelCount, "loader");
+            loadService = ExecutorUtil.newFixedThreadPool(parallelThreads, "loader");
             List<InputTaskItem> taskItems = prepareTaskItems(structs, scatter);
             List<CompletableFuture<Void>> loadTasks = new ArrayList<>();
 

hugegraph-loader/src/main/java/org/apache/hugegraph/loader/executor/LoadOptions.java

@@ -45,6 +45,8 @@ public final class LoadOptions implements Cloneable {
     public static final String HTTPS_SCHEMA = "https";
     public static final String HTTP_SCHEMA = "http";
     private static final int CPUS = Runtime.getRuntime().availableProcessors();
+    private static final int DEFAULT_MAX_CONNECTIONS = CPUS * 4;
+    private static final int DEFAULT_MAX_CONNECTIONS_PER_ROUTE = CPUS * 2;
     private static final int MINIMUM_REQUIRED_ARGS = 3;
 
     @Parameter(names = {"-f", "--file"}, required = true, arity = 1,
@@ -156,7 +158,9 @@ public final class LoadOptions implements Cloneable {
 
     @Parameter(names = {"--batch-insert-threads"}, arity = 1,
                validateWith = {PositiveValidator.class},
-               description = "The number of threads to execute batch insert")
+               description = "The number of threads to execute batch insert. " +
+                             "If max-conn/max-conn-per-route keep defaults, " +
+                             "they may be auto-adjusted based on this value")
     public int batchInsertThreads = CPUS;
 
     @Parameter(names = {"--single-insert-threads"}, arity = 1,
@@ -165,21 +169,30 @@ public final class LoadOptions implements Cloneable {
     public int singleInsertThreads = 8;
 
     @Parameter(names = {"--max-conn"}, arity = 1,
-               description = "Max number of HTTP connections to server")
-    public int maxConnections = CPUS * 4;
+               validateWith = {PositiveValidator.class},
+               description = "Max number of HTTP connections to server. " +
+                             "If left as default and batch-insert-threads is " +
+                             "set, this may be auto-adjusted")
+    public int maxConnections = DEFAULT_MAX_CONNECTIONS;
 
     @Parameter(names = {"--max-conn-per-route"}, arity = 1,
-               description = "Max number of HTTP connections to each route")
-    public int maxConnectionsPerRoute = CPUS * 2;
+               validateWith = {PositiveValidator.class},
+               description = "Max number of HTTP connections to each route. " +
+                             "If left as default and batch-insert-threads is " +
+                             "set, this may be auto-adjusted")
+    public int maxConnectionsPerRoute = DEFAULT_MAX_CONNECTIONS_PER_ROUTE;
 
     @Parameter(names = {"--batch-size"}, arity = 1,
                validateWith = {PositiveValidator.class},
                description = "The number of lines in each submit")
     public int batchSize = 500;
 
-    @Parameter(names = {"--parallel-count"}, arity = 1,
-            description = "The number of parallel read pipelines")
-    public int parallelCount = 1;
+    @Parameter(names = {"--parallel-count", "--parser-threads"}, arity = 1,
+               validateWith = {PositiveValidator.class},
+               description = "The number of parallel read pipelines. " +
+                             "Default: auto max(2, cpu/2). " +
+                             "Must be >= 1")
+    public int parallelThreads = Math.max(2, CPUS/2);
 
     @Parameter(names = {"--start-file"}, arity = 1,
             description = "start file index for partial loading")

hugegraph-loader/src/main/java/org/apache/hugegraph/loader/task/TaskManager.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.hugegraph.loader.util.Printer;
 import org.slf4j.Logger;
 
 import org.apache.hugegraph.loader.builder.Record;
@@ -137,6 +138,9 @@ public void shutdown() {
 
     public void submitBatch(InputStruct struct, ElementMapping mapping,
                             List<Record> batch) {
+        if (this.context.stopped()) {
+            return;
+        }
         long start = System.currentTimeMillis();
         try {
             this.batchSemaphore.acquire();
@@ -152,13 +156,12 @@ public void submitBatch(InputStruct struct, ElementMapping mapping,
         CompletableFuture.runAsync(task, this.batchService).whenComplete(
             (r, e) -> {
                 if (e != null) {
-                    LOG.warn("Batch insert {} error, try single insert",
-                             mapping.type(), e);
-                    // The time of single insert is counted separately
-                    this.submitInSingle(struct, mapping, batch);
-                } else {
-                    summary.metrics(struct).minusFlighting(batch.size());
+                    LOG.error("Batch insert {} error, interrupting import", mapping.type(), e);
+                    this.context.stopLoading();
+                    Printer.printError("Batch insert %s failed, stop loading, Please check the logs",
+                                       mapping.type().string());
                 }
+                summary.metrics(struct).minusFlighting(batch.size());
 
                 this.batchSemaphore.release();
                 long end = System.currentTimeMillis();