anph
/
hugegraph-sync
forked from hugegraph/hugegraph-sync
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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
This commit is contained in:
zhangyi51 2018-10-30 15:22:14 +08:00 committed by Linary
parent 15e9fc6008
commit c02ed7107c
4 changed files with 21 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
hugegraph-core/src/main/java/com/baidu/hugegraph/HugeGraph.java
View File

@ -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 {

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

@ -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();

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

@ -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 {

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

@ -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);
}
}
}