Revert "fix:apache#488 bypass server for hbase bulk delete massive ve…

…rtices and edges" This reverts commit d8b31fc
haohao0103 · Aug 2, 2023 · b0918d5 · b0918d5
1 parent e610fb7
commit b0918d5
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 31 deletions.
diff --git a/hugegraph-loader/assembly/static/bin/get-params.sh b/hugegraph-loader/assembly/static/bin/get-params.sh
@@ -27,8 +27,7 @@ function get_params() {
       --incremental-mode | --failure-mode | --batch-insert-threads | --single-insert-threads | \
       --max-conn | --max-conn-per-route | --batch-size | --max-parse-errors | --max-insert-errors | \
       --timeout | --shutdown-timeout | --retry-times | --retry-interval | --check-vertex | \
-      --print-progress | --dry-run | --sink-type | --vertex-partitions | --edge-partitions |  \
-      --action-type | --help )
+      --print-progress | --dry-run | --sink-type | --vertex-partitions | --edge-partitions | --help )
         HUGEGRAPH_PARAMS="$HUGEGRAPH_PARAMS $1 $2"
         shift 2
         ;;

diff --git a/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/constant/Constants.java b/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/constant/Constants.java
@@ -81,5 +81,4 @@ public final class Constants {
     public static final String LOAD_DATA_PARSE_SUFFIX = "parse";
     public static final String LOAD_DATA_SER_SUFFIX = "ser";
     public static final String LOAD_DATA_INSERT_SUFFIX = "insert";
-    public static final String DELETE_ACTION = "delete";
 }
diff --git a/...aph-loader/src/main/java/org/apache/hugegraph/loader/direct/loader/HBaseDirectLoader.java b/...aph-loader/src/main/java/org/apache/hugegraph/loader/direct/loader/HBaseDirectLoader.java
@@ -26,7 +26,6 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FsShell;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
@@ -201,29 +200,23 @@ List<Tuple2<ImmutableBytesWritable, KeyValue>> buildAndSer(HBaseSerializer seria
                                                            struct.input().type()));
             }
 
-            if (elementsElement != null) {
-                for (GraphElement graphElement : elementsElement) {
-                    final byte[] rowkey = serializer.getKeyBytes(graphElement);
-                    final byte[] values = serializer.getValueBytes(graphElement);
-                    ImmutableBytesWritable rowKey = new ImmutableBytesWritable();
-                    rowKey.set(rowkey);
-                    if (loadOptions.actionType.equals(Constants.DELETE_ACTION)) {
-                        KeyValue  keyValue = new KeyValue(
-                                rowkey,Bytes.toBytes(Constants.HBASE_COL_FAMILY),
-                                Bytes.toBytes(Constants.EMPTY_STR),HConstants.LATEST_TIMESTAMP,
-                                KeyValue.Type.DeleteFamily);
-                        Tuple2<ImmutableBytesWritable, KeyValue> tuple2 =
-                                new Tuple2<>(rowKey,keyValue);
-                        result.add(tuple2);
-                    } else {
-                        KeyValue  keyValue = new KeyValue(rowkey,
-                                Bytes.toBytes(Constants.HBASE_COL_FAMILY),
-                                Bytes.toBytes(Constants.EMPTY_STR),
-                                values);
-                        Tuple2<ImmutableBytesWritable, KeyValue> tuple2 =
-                                new Tuple2<>(rowKey,keyValue);
-                        result.add(tuple2);
-                    }
+            boolean isVertex = builder.mapping().type().isVertex();
+            if (isVertex) {
+                for (Vertex vertex : (List<Vertex>) (Object) elementsElement) {
+                    LOG.debug("vertex already build done {} ", vertex.toString());
+                    Tuple2<ImmutableBytesWritable, KeyValue> tuple2 =
+                            vertexSerialize(serializer, vertex);
+                    loadDistributeMetrics.increaseDisVertexInsertSuccess(builder.mapping());
+                    result.add(tuple2);
+                }
+            } else {
+                for (Edge edge : (List<Edge>) (Object) elementsElement) {
+                    LOG.debug("edge already build done {}", edge.toString());
+                    Tuple2<ImmutableBytesWritable, KeyValue> tuple2 =
+                            edgeSerialize(serializer, edge);
+                    loadDistributeMetrics.increaseDisEdgeInsertSuccess(builder.mapping());
+                    result.add(tuple2);
+
                 }
             }
         }

diff --git a/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/executor/LoadOptions.java b/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/executor/LoadOptions.java
@@ -233,10 +233,6 @@ public class LoadOptions implements Serializable {
                description = "hbaseZKParent")
     public String hbaseZKParent;
 
-    @Parameter(names = {"--action-type"},  arity = 1,
-            description = "add/update or delete")
-    public String actionType="add";
-
     public String workModeString() {
         if (this.incrementalMode) {
             return "INCREMENTAL MODE";