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

fix bug: remove secondary index of cassandra backend exceeding 65535 limit (#387) 

fixed: #384

Change-Id: I3b0db265b621c149404a97ab688af8bd2e4ca4bd
This commit is contained in:
zhoney 2019-03-08 19:57:12 +08:00 committed by Linary
parent 6eaeee81d8
commit f4031f1e05
3 changed files with 23 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTables.java
View File

@ -558,12 +558,18 @@ public class CassandraTables {
}
final String FIELD_VALUES = formatKey(HugeKeys.FIELD_VALUES);
int count = 0;
for (Iterator<Row> it = rs.iterator(); it.hasNext();) {
fieldValues = it.next().get(FIELD_VALUES, String.class);
Delete delete = QueryBuilder.delete().from(this.table());
delete.where(formatEQ(HugeKeys.INDEX_LABEL_ID, indexLabel));
delete.where(formatEQ(HugeKeys.FIELD_VALUES, fieldValues));
session.add(delete);
if (++count >= COMMIT_DELETE_BATCH) {
session.commit();
count = 0;
}
}
}

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

@ -1324,7 +1324,7 @@ public class GraphTransaction extends IndexableTransaction {
this.traverseVerticesByLabel(vertexLabel, vertex -> {
this.removeVertex((HugeVertex) vertex);
this.commitIfGtSize(COMMIT_BATCH);
});
}, true);
this.commit();
} catch (Exception e) {
LOG.error("Failed to remove vertices", e);
@ -1354,7 +1354,7 @@ public class GraphTransaction extends IndexableTransaction {
this.traverseEdgesByLabel(edgeLabel, edge -> {
this.removeEdge((HugeEdge) edge);
this.commitIfGtSize(COMMIT_BATCH);
});
}, true);
}
this.commit();
} catch (Exception e) {
@ -1366,18 +1366,19 @@ public class GraphTransaction extends IndexableTransaction {
}
public void traverseVerticesByLabel(VertexLabel label,
Consumer<Vertex> consumer) {
this.traverseByLabel(label, this::queryVertices, consumer);
Consumer<Vertex> consumer,
boolean remove) {
this.traverseByLabel(label, this::queryVertices, consumer, remove);
}
public void traverseEdgesByLabel(EdgeLabel label,
Consumer<Edge> consumer) {
this.traverseByLabel(label, this::queryEdges, consumer);
public void traverseEdgesByLabel(EdgeLabel label, Consumer<Edge> consumer,
boolean remove) {
this.traverseByLabel(label, this::queryEdges, consumer, remove);
}
private <T> void traverseByLabel(SchemaLabel label,
Function<Query, Iterator<T>> fetcher,
Consumer<T> consumer) {
Consumer<T> consumer, boolean remove) {
HugeType type = label.type() == HugeType.VERTEX_LABEL ?
HugeType.VERTEX : HugeType.EDGE;
ConditionQuery query = new ConditionQuery(type);
@ -1408,9 +1409,11 @@ public class GraphTransaction extends IndexableTransaction {
query.capacity(Query.NO_CAPACITY);
query.eq(HugeKeys.LABEL, label.id());
int pass = 0;
int counter = 0;
int counter;
do {
query.offset(pass++ * Query.DEFAULT_CAPACITY);
if (!remove) {
query.offset(pass++ * Query.DEFAULT_CAPACITY);
}
// Process every element in current batch
Iterator<T> itor = fetcher.apply(query);
for (counter = 0; itor.hasNext(); ++counter) {

6
hugegraph-core/src/main/java/com/baidu/hugegraph/job/schema/RebuildIndexCallable.java
View File

@ -123,12 +123,14 @@ public class RebuildIndexCallable extends SchemaCallable {
if (label.type() == HugeType.VERTEX_LABEL) {
@SuppressWarnings("unchecked")
Consumer<Vertex> consumer = (Consumer<Vertex>) indexUpdater;
graphTx.traverseVerticesByLabel((VertexLabel) label, consumer);
graphTx.traverseVerticesByLabel((VertexLabel) label,
consumer, false);
} else {
assert label.type() == HugeType.EDGE_LABEL;
@SuppressWarnings("unchecked")
Consumer<Edge> consumer = (Consumer<Edge>) indexUpdater;
graphTx.traverseEdgesByLabel((EdgeLabel) label, consumer);
graphTx.traverseEdgesByLabel((EdgeLabel) label,
consumer, false);
}
graphTx.commit();