Commit index updates in batch

To avoid commit all together during rebuilding index,
especially for Cassandra backend, which has batch limit 65535

fixed: #144
implemented: #82

Change-Id: I88ff4bc878bc24122f0bb6ecf9964246a083b9ab

README.md

@@ -1,8 +1,8 @@
 # HugeGraph
 
 [![License](https://img.shields.io/badge/license-Apache%202-0E78BA.svg)](https://www.apache.org/licenses/LICENSE-2.0.html)
-[![Build Status](https://travis-ci.org/hugegraph/hugegraph.svg?branch=master)](https://travis-ci.org/hugegraph/hugegraph)
-[![codecov](https://codecov.io/gh/hugegraph/hugegraph/branch/master/graph/badge.svg)](https://codecov.io/gh/hugegraph/hugegraph)
+[![Build Status](https://travis-ci.org/hugegraph/hugegraph.svg?branch=release-0.7)](https://travis-ci.org/hugegraph/hugegraph)
+[![codecov](https://codecov.io/gh/hugegraph/hugegraph/branch/release-0.7/graph/badge.svg)](https://codecov.io/gh/hugegraph/hugegraph)
 
 HugeGraph is a fast-speed and highly-scalable [graph database](https://en.wikipedia.org/wiki/Graph_database). Billions of vertices and edges can be easily stored into and queried from HugeGraph due to its excellent OLTP ability. As compliance to [Apache TinkerPop 3](https://tinkerpop.apache.org/) framework, various complicated graph queries can be accomplished through [Gremlin](https://tinkerpop.apache.org/gremlin.html)(a powerful graph traversal language).
 

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

@@ -550,6 +550,12 @@ public class HugeGraph implements Graph {
             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/tx/AbstractTransaction.java

@@ -170,6 +170,12 @@ public abstract class AbstractTransaction implements Transaction {
         }
     }
 
+    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

@@ -77,6 +77,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;
 
@@ -1111,6 +1112,9 @@ public class GraphIndexTransaction extends AbstractTransaction {
                     HugeVertex vertex = (HugeVertex) itor.next();
                     for (Id id : indexLabelIds) {
                         this.updateIndex(id, vertex, false);
+                        // Commit per small batch to avoid too much data
+                        // in single commit, especially for Cassandra backend.
+                        this.commitIfGtSize(REBUILD_COMMIT_BATCH);
                     }
                 }
             } else {
@@ -1123,6 +1127,9 @@ public class GraphIndexTransaction extends AbstractTransaction {
                     HugeEdge edge = (HugeEdge) itor.next();
                     for (Id id : indexLabelIds) {
                         this.updateIndex(id, edge, false);
+                        // Commit per small batch to avoid too much data
+                        // in single commit, especially for Cassandra backend.
+                        this.commitIfGtSize(REBUILD_COMMIT_BATCH);
                     }
                 }
             }

hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBSessions.java

@@ -19,14 +19,10 @@
 
 package com.baidu.hugegraph.backend.store.rocksdb;
 
-import java.io.IOException;
-import java.nio.file.Paths;
 import java.util.Set;
 
-import org.apache.commons.io.FileUtils;
 import org.rocksdb.RocksDBException;
 
-import com.baidu.hugegraph.backend.BackendException;
 import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumnIterator;
 import com.baidu.hugegraph.backend.store.BackendSession;
 import com.baidu.hugegraph.backend.store.BackendSessionPool;
@@ -48,21 +44,6 @@ public abstract class RocksDBSessions extends BackendSessionPool {
     @Override
     public abstract Session session();
 
-    public String wrapPath(String path) {
-        return wrapPath(path, this.store);
-    }
-
-    public static String wrapPath(String path, String store) {
-        // Ensure the `path` exists
-        try {
-            FileUtils.forceMkdir(FileUtils.getFile(path));
-        } catch (IOException e) {
-            throw new BackendException(e.getMessage(), e);
-        }
-        // Join with store type
-        return Paths.get(path, store).toString();
-    }
-
     /**
      * Session for RocksDB
      */

hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStdSessions.java

@@ -62,15 +62,13 @@ public class RocksDBStdSessions extends RocksDBSessions {
     private final HugeConfig conf;
     private final RocksDB rocksdb;
 
-    public RocksDBStdSessions(HugeConfig config, String database, String store)
+    public RocksDBStdSessions(HugeConfig config, String dataPath,
+                              String walPath, String database, String store)
                               throws RocksDBException {
         super(database, store);
 
         this.conf = config;
 
-        String dataPath = wrapPath(this.conf.get(RocksDBOptions.DATA_PATH));
-        String walPath = wrapPath(this.conf.get(RocksDBOptions.WAL_PATH));
-
         // Init options
         Options options = new Options();
         RocksDBStdSessions.initOptions(this.conf, options, options, options);
@@ -83,14 +81,12 @@ public class RocksDBStdSessions extends RocksDBSessions {
         this.rocksdb = RocksDB.open(options, dataPath);
     }
 
-    public RocksDBStdSessions(HugeConfig config, String database, String store,
+    public RocksDBStdSessions(HugeConfig config, String dataPath,
+                              String walPath, String database, String store,
                               List<String> cfNames) throws RocksDBException {
         super(database, store);
         this.conf = config;
 
-        String dataPath = wrapPath(this.conf.get(RocksDBOptions.DATA_PATH));
-        String walPath = wrapPath(this.conf.get(RocksDBOptions.WAL_PATH));
-
         // Old CFs should always be opened
         List<String> cfs = this.mergeOldCFs(dataPath, cfNames);
 

hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java

@@ -19,6 +19,8 @@
 
 package com.baidu.hugegraph.backend.store.rocksdb;
 
+import java.io.IOException;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -29,6 +31,7 @@ import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
+import org.apache.commons.io.FileUtils;
 import org.rocksdb.RocksDBException;
 import org.slf4j.Logger;
 
@@ -129,9 +132,10 @@ public abstract class RocksDBStore implements BackendStore {
         }
 
         // Open base disk
-        String dataPath = config.get(RocksDBOptions.DATA_PATH);
-        dataPath = RocksDBSessions.wrapPath(dataPath, this.store);
-        this.sessions = this.open(config, dataPath, this.tableNames());
+        String dataPath = this.wrapPath(config.get(RocksDBOptions.DATA_PATH));
+        String walPath = this.wrapPath(config.get(RocksDBOptions.WAL_PATH));
+
+        this.sessions = this.open(config, dataPath, walPath, this.tableNames());
 
         // Open tables with optimized disk
         List<String> disks = config.get(RocksDBOptions.DATA_DISKS);
@@ -140,18 +144,19 @@ public abstract class RocksDBStore implements BackendStore {
             for (Entry<HugeType, String> e : this.tableDiskMapping.entrySet()) {
                 String table = this.table(e.getKey()).table();
                 String disk = e.getValue();
-                this.open(config, disk, Arrays.asList(table));
+                this.open(config, disk, disk, Arrays.asList(table));
             }
         }
     }
 
     protected RocksDBSessions open(HugeConfig config, String dataPath,
-                                   List<String> tableNames) {
+                                   String walPath, List<String> tableNames) {
         LOG.info("Opening RocksDB with data path: {}", dataPath);
 
         RocksDBSessions sessions = null;
         try {
-            sessions = this.openSessionPool(config, tableNames);
+            sessions = this.openSessionPool(config, dataPath,
+                                            walPath, tableNames);
         } catch (RocksDBException e) {
             if (dbs.containsKey(dataPath)) {
                 if (e.getMessage().contains("No locks available")) {
@@ -162,7 +167,8 @@ public abstract class RocksDBStore implements BackendStore {
                     try {
                         // Will open old CFs(of other keyspace)
                         final List<String> none = ImmutableList.of();
-                        sessions = this.openSessionPool(config, none);
+                        sessions = this.openSessionPool(config, dataPath,
+                                                        walPath, none);
                     } catch (RocksDBException e1) {
                         // Let it throw later
                         e = e1;
@@ -174,7 +180,8 @@ public abstract class RocksDBStore implements BackendStore {
                          "try to init CF later", dataPath, this.database);
                 try {
                     // Only open default CF, won't open old CFs
-                    sessions = this.openSessionPool(config, null);
+                    sessions = this.openSessionPool(config, dataPath,
+                                                    walPath, null);
                 } catch (RocksDBException e1) {
                     LOG.error("Failed to open RocksDB with default CF", e1);
                 }
@@ -198,12 +205,15 @@ public abstract class RocksDBStore implements BackendStore {
     }
 
     protected RocksDBSessions openSessionPool(HugeConfig config,
+                                              String dataPath, String walPath,
                                               List<String> tableNames)
                                               throws RocksDBException {
         if (tableNames == null) {
-            return new RocksDBStdSessions(config, this.database, this.store);
+            return new RocksDBStdSessions(config, dataPath, walPath,
+                                          this.database, this.store);
         } else {
-            return new RocksDBStdSessions(config, this.database, this.store,
+            return new RocksDBStdSessions(config, dataPath, walPath,
+                                          this.database, this.store,
                                           tableNames);
         }
     }
@@ -401,7 +411,7 @@ public abstract class RocksDBStore implements BackendStore {
             String store = pair[0].trim();
             HugeType table = HugeType.valueOf(pair[1].trim().toUpperCase());
             if (this.store.equals(store)) {
-                path = RocksDBSessions.wrapPath(path, this.store);
+                path = this.wrapPath(path);
                 this.tableDiskMapping.put(table, path);
             }
         }
@@ -414,6 +424,17 @@ public abstract class RocksDBStore implements BackendStore {
         return db;
     }
 
+    protected String wrapPath(String path) {
+        // Ensure the `path` exists
+        try {
+            FileUtils.forceMkdir(FileUtils.getFile(path));
+        } catch (IOException e) {
+            throw new BackendException(e.getMessage(), e);
+        }
+        // Join with store type
+        return Paths.get(path, this.store).toString();
+    }
+
     /***************************** Store defines *****************************/
 
     public static class RocksDBSchemaStore extends RocksDBStore {

hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstSessions.java

@@ -22,12 +22,8 @@ package com.baidu.hugegraph.backend.store.rocksdbsst;
 import java.io.File;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.commons.lang3.tuple.Pair;
@@ -38,7 +34,6 @@ import org.rocksdb.SstFileWriter;
 
 import com.baidu.hugegraph.backend.BackendException;
 import com.baidu.hugegraph.backend.store.BackendEntry.BackendColumnIterator;
-import com.baidu.hugegraph.backend.store.rocksdb.RocksDBOptions;
 import com.baidu.hugegraph.backend.store.rocksdb.RocksDBSessions;
 import com.baidu.hugegraph.backend.store.rocksdb.RocksDBStdSessions;
 import com.baidu.hugegraph.config.HugeConfig;
@@ -51,11 +46,12 @@ public class RocksDBSstSessions extends RocksDBSessions {
     private final String dataPath;
     private final Map<String, SstFileWriter> tables;
 
-    public RocksDBSstSessions(HugeConfig conf, String database, String store) {
+    public RocksDBSstSessions(HugeConfig conf, String dataPath,
+                              String database, String store) {
         super(database, store);
 
         this.conf = conf;
-        this.dataPath = this.wrapPath(this.conf.get(RocksDBOptions.DATA_PATH));
+        this.dataPath = dataPath;
         this.tables = new ConcurrentHashMap<>();
 
         File path = new File(this.dataPath);
@@ -64,9 +60,10 @@ public class RocksDBSstSessions extends RocksDBSessions {
         }
     }
 
-    public RocksDBSstSessions(HugeConfig config, String database, String store,
+    public RocksDBSstSessions(HugeConfig config, String dataPath,
+                              String database, String store,
                               List<String> tableNames) throws RocksDBException {
-        this(config, database, store);
+        this(config, dataPath, database, store);
         for (String table : tableNames) {
             this.createTable(table);
         }

hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstStore.java

@@ -40,13 +40,14 @@ public abstract class RocksDBSstStore extends RocksDBStore {
 
     @Override
     protected RocksDBSessions openSessionPool(HugeConfig config,
+                                              String dataPath, String walPath,
                                               List<String> tableNames)
                                               throws RocksDBException {
         if (tableNames == null) {
-            return new RocksDBSstSessions(config, this.database(),
+            return new RocksDBSstSessions(config, dataPath, this.database(),
                                           this.store());
         } else {
-            return new RocksDBSstSessions(config, this.database(),
+            return new RocksDBSstSessions(config, dataPath, this.database(),
                                           this.store(), tableNames);
         }
     }

hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBStoreProvider.java

@@ -114,7 +114,7 @@ public class ScyllaDBStoreProvider extends CassandraStoreProvider {
                 registerTableManager(HugeType.EDGE_IN,
                                      ScyllaDBTablesWithMV.Edge.in(store));
             } else {
-                registerTableManager(HugeType.EDGE_OUT,
+                registerTableManager(HugeType.VERTEX,
                                      new ScyllaDBTables.Vertex(store));
                 registerTableManager(HugeType.EDGE_OUT,
                                      ScyllaDBTables.Edge.out(store));

hugegraph-test/pom.xml

@@ -155,9 +155,6 @@
         <resources>
             <resource>
                 <directory>src/main/resources/</directory>
-                <includes>
-                    <include>hugegraph.properties</include>
-                </includes>
                 <filtering>true</filtering>
             </resource>
         </resources>

hugegraph-test/src/main/java/com/baidu/hugegraph/tinkerpop/TestGraph.java

@@ -39,10 +39,10 @@ import com.baidu.hugegraph.schema.SchemaManager;
 import com.baidu.hugegraph.structure.HugeFeatures;
 import com.baidu.hugegraph.type.define.IdStrategy;
 
-@Graph.OptIn("com.baidu.hugegraph.tinkerpop.StructureStandardTest")
-@Graph.OptIn("com.baidu.hugegraph.tinkerpop.ProcessStandardTest")
-@Graph.OptIn("com.baidu.hugegraph.tinkerpop.StructurePerformanceTest")
-@Graph.OptIn("com.baidu.hugegraph.tinkerpop.ProcessPerformanceTest")
+@Graph.OptIn("com.baidu.hugegraph.tinkerpop.StructureBasicSuite")
+@Graph.OptIn("com.baidu.hugegraph.tinkerpop.ProcessBasicSuite")
+@Graph.OptIn("com.baidu.hugegraph.tinkerpop.StructurePerformanceSuite")
+@Graph.OptIn("com.baidu.hugegraph.tinkerpop.ProcessPerformanceSuite")
 public class TestGraph implements Graph {
 
     public static final String DEFAULT_VL = "vertex";

hugegraph-test/src/main/java/com/baidu/hugegraph/tinkerpop/TestGraphProvider.java

@@ -23,6 +23,7 @@ import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
+import java.net.URL;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
@@ -95,9 +96,12 @@ public class TestGraphProvider extends AbstractGraphProvider {
             filter = DEFAULT_FILTER;
         }
 
-        String blackList = TestGraphProvider.class.getClassLoader()
-                           .getResource(filter).getPath();
-        File file = new File(blackList);
+        URL blackList = TestGraphProvider.class.getClassLoader()
+                                               .getResource(filter);
+        E.checkArgument(blackList != null,
+                        "Can't find tests filter '%s' in resource directory",
+                        filter);
+        File file = new File(blackList.getPath());
         E.checkArgument(
                 file.exists() && file.isFile() && file.canRead(),
                 "Need to specify a readable filter file rather than: %s",

hugegraph-test/src/main/java/com/baidu/hugegraph/unit/rocksdb/BaseRocksDBUnitTest.java

@@ -34,7 +34,6 @@ import org.junit.Before;
 import org.mockito.Mockito;
 import org.rocksdb.RocksDBException;
 
-import com.baidu.hugegraph.backend.store.rocksdb.RocksDBOptions;
 import com.baidu.hugegraph.backend.store.rocksdb.RocksDBSessions;
 import com.baidu.hugegraph.backend.store.rocksdb.RocksDBStdSessions;
 import com.baidu.hugegraph.config.HugeConfig;
@@ -117,9 +116,8 @@ public class BaseRocksDBUnitTest extends BaseUnitTest {
         Configuration conf = Mockito.mock(PropertiesConfiguration.class);
         Mockito.when(conf.getKeys()).thenReturn(Collections.emptyIterator());
         HugeConfig config = new HugeConfig(conf);
-        config.setProperty(RocksDBOptions.DATA_PATH.name(), DB_PATH);
-        config.setProperty(RocksDBOptions.WAL_PATH.name(), DB_PATH);
-        RocksDBSessions rocks = new RocksDBStdSessions(config, "db", "store");
+        RocksDBSessions rocks = new RocksDBStdSessions(config, DB_PATH, DB_PATH,
+                                                       "db", "store");
         rocks.createTable(table);
         return rocks;
     }

hugegraph-test/src/main/resources/hugegraph.properties

@@ -5,8 +5,8 @@ serializer=${serializer}
 
 store=hugegraph
 
-vertex.tx_capacity=1000
-edge.tx_capacity=1000
+vertex.tx_capacity=10000
+edge.tx_capacity=10000
 
 # cassandra backend config
 cassandra.host=127.0.0.1