Commit index updates in batch

hugegraph-api/src/main/java/com/baidu/hugegraph/api/job/GremlinAPI.java

@@ -75,10 +75,11 @@ public Map<String, Id> post(@Context GraphManager manager,
 
         HugeGraph g = graph(manager, graph);
         request.aliase(graph, "graph");
-        JobBuilder builder = JobBuilder.of(g).name(request.name())
-                                       .input(request.toJson())
-                                       .job(new GremlinJob());
-        return ImmutableMap.of("task_id", builder.schedule());
+        JobBuilder<Object> builder = JobBuilder.of(g);
+        builder.name(request.name())
+               .input(request.toJson())
+               .job(new GremlinJob());
+        return ImmutableMap.of("task_id", builder.schedule().id());
     }
 
     public static class GremlinJob extends Job<Object> {

hugegraph-core/src/main/java/com/baidu/hugegraph/HugeGraph.java

@@ -550,6 +550,12 @@ public boolean closed() {
             return this.refs.get() == 0;
         }
 
+        public void commitIfGtSize(int size) {
+            // Only committing graph transaction data if reaching batch size
+            // is OK, bacause schema transaction is auto committed.
+            this.graphTransaction().commitIfGtSize(size);
+        }
+
         @Override
         public void commit() {
             try {

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/Transaction.java

@@ -23,6 +23,8 @@ public interface Transaction {
 
     public void commit() throws BackendException;
 
+    public void commitIfGtSize(int size) throws BackendException;
+
     public void rollback() throws BackendException;
 
     public boolean autoCommit();

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CacheManager.java

@@ -87,11 +87,9 @@ public Cache cache(String name) {
     }
 
     public Cache cache(String name, int capacity) {
-        Cache cache = this.caches.get(name);
-        if (cache == null) {
-            cache = new RamCache(capacity);
-            this.caches.put(name, cache);
+        if (!this.caches.containsKey(name)) {
+            this.caches.putIfAbsent(name, new RamCache(capacity));
         }
-        return cache;
+        return this.caches.get(name);
     }
 }

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/AbstractTransaction.java

@@ -170,6 +170,12 @@ public void commit() throws BackendException {
         }
     }
 
+    public void commitIfGtSize(int size) throws BackendException {
+        if (this.mutationSize() >= size) {
+            this.commit();
+        }
+    }
+
     @Watched(prefix = "tx")
     @Override
     public void rollback() throws BackendException {

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphIndexTransaction.java

@@ -76,6 +76,7 @@ public class GraphIndexTransaction extends AbstractTransaction {
 
     private static final String INDEX_EMPTY_SYM = "\u0000";
     private static final Query EMPTY_QUERY = new ConditionQuery(null);
+    private static final int REBUILD_COMMIT_BATCH = 1000;
 
     private final Analyzer textAnalyzer;
 
@@ -1100,30 +1101,13 @@ public void rebuildIndex(HugeType type, Id label,
              * They have different id lead to it can't compare and optimize
              */
             this.commit();
-
             if (type == HugeType.VERTEX_LABEL) {
-                ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
-                query.eq(HugeKeys.LABEL, label);
-                query.capacity(Query.NO_CAPACITY);
-                for (Iterator<Vertex> itor = graphTx.queryVertices(query);
-                     itor.hasNext();) {
-                    HugeVertex vertex = (HugeVertex) itor.next();
-                    for (Id id : indexLabelIds) {
-                        this.updateIndex(id, vertex, false);
-                    }
-                }
+                this.rebuildIndex(HugeType.VERTEX, label, indexLabelIds,
+                                  graphTx::queryVerticesFromBackend);
             } else {
                 assert type == HugeType.EDGE_LABEL;
-                ConditionQuery query = new ConditionQuery(HugeType.EDGE);
-                query.eq(HugeKeys.LABEL, label);
-                query.capacity(Query.NO_CAPACITY);
-                for (Iterator<Edge> itor = graphTx.queryEdges(query);
-                     itor.hasNext();) {
-                    HugeEdge edge = (HugeEdge) itor.next();
-                    for (Id id : indexLabelIds) {
-                        this.updateIndex(id, edge, false);
-                    }
-                }
+                this.rebuildIndex(HugeType.EDGE, label, indexLabelIds,
+                                  graphTx::queryEdgesFromBackend);
             }
             this.commit();
 
@@ -1136,6 +1120,34 @@ public void rebuildIndex(HugeType type, Id label,
         }
     }
 
+    private <T> void rebuildIndex(HugeType type, Id label,
+                                  Collection<Id> indexLabelIds,
+                                  Function<Query, Iterator<T>> fetcher) {
+        assert type == HugeType.VERTEX || type == HugeType.EDGE;
+        ConditionQuery query = new ConditionQuery(type);
+        query.eq(HugeKeys.LABEL, label);
+        query.limit(Query.DEFAULT_CAPACITY);
+
+        int pass = 0;
+        int counter;
+        do {
+            counter = 0;
+            query.offset(pass++ * Query.DEFAULT_CAPACITY);
+            Iterator<T> itor = fetcher.apply(query);
+            while (itor.hasNext()) {
+                HugeElement element = (HugeElement) itor.next();
+                for (Id id : indexLabelIds) {
+                    this.updateIndex(id, element, false);
+                    // Commit per small batch to avoid too much data
+                    // in single commit, especially for Cassandra backend.
+                    this.commitIfGtSize(REBUILD_COMMIT_BATCH);
+                }
+                ++counter;
+            }
+            assert counter <= Query.DEFAULT_CAPACITY;
+        } while (counter == Query.DEFAULT_CAPACITY);
+    }
+
     private static class MatchedLabel {
 
         private SchemaLabel schemaLabel;

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphTransaction.java

@@ -446,7 +446,7 @@ public Iterator<Vertex> queryVertices(Query query) {
         return r;
     }
 
-    private Iterator<HugeVertex> queryVerticesFromBackend(Query query) {
+    protected Iterator<HugeVertex> queryVerticesFromBackend(Query query) {
         assert query.resultType().isVertex();
 
         Iterator<BackendEntry> entries = this.query(query);
@@ -562,7 +562,7 @@ public Iterator<Edge> queryEdges(Query query) {
         return r;
     }
 
-    private Iterator<HugeEdge> queryEdgesFromBackend(Query query) {
+    protected Iterator<HugeEdge> queryEdgesFromBackend(Query query) {
         assert query.resultType().isEdge();
 
         Iterator<BackendEntry> entries = this.query(query);