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

HugeGraph-387: add async-task framework 

Change-Id: I452f3a1c9a7ccf1580075e9fc18456b1b4aae1c6
This commit is contained in:
Zhangmei Li 2018-03-14 18:43:07 +08:00 committed by liningrui
parent fed5f76b4f
commit 57875977ec
49 changed files with 1308 additions and 210 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hugegraph-api/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

2
hugegraph-cassandra/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

18
hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraSessionPool.java
View File

@ -46,7 +46,8 @@ public class CassandraSessionPool extends BackendSessionPool {
private Cluster cluster;
private String keyspace;
public CassandraSessionPool(String keyspace) {
public CassandraSessionPool(String keyspace, String store) {
super(keyspace + "/" + store);
this.cluster = null;
this.keyspace = keyspace;
}
@ -149,9 +150,6 @@ public class CassandraSessionPool extends BackendSessionPool {
public Session() {
this.session = null;
this.batch = new BatchStatement(); // LOGGED
try {
this.open();
} catch (InvalidQueryException ignored) {}
}
public BatchStatement add(Statement statement) {
@ -212,16 +210,24 @@ public class CassandraSessionPool extends BackendSessionPool {
return this.session.execute(statement, args);
}
private void tryOpen() {
assert this.session == null;
try {
this.open();
} catch (InvalidQueryException ignored) {}
}
public void open() {
assert this.session == null;
this.session = cluster().connect(keyspace());
}
@Override
public boolean closed() {
if (this.session == null) {
return true;
this.tryOpen();
}
return this.session.isClosed();
return this.session == null ? true : this.session.isClosed();
}
@Override

13
hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraStore.java
View File

@ -37,7 +37,6 @@ import com.baidu.hugegraph.backend.store.BackendStore;
import com.baidu.hugegraph.backend.store.BackendStoreProvider;
import com.baidu.hugegraph.config.HugeConfig;
import com.baidu.hugegraph.type.HugeType;
import com.baidu.hugegraph.type.define.Directions;
import com.baidu.hugegraph.util.E;
import com.baidu.hugegraph.util.Log;
import com.datastax.driver.core.Cluster;
@ -73,7 +72,7 @@ public abstract class CassandraStore implements BackendStore {
this.keyspace = keyspace;
this.store = store;
this.sessions = new CassandraSessionPool(this.keyspace);
this.sessions = new CassandraSessionPool(keyspace, store);
this.tables = new ConcurrentHashMap<>();
this.conf = null;
@ -495,17 +494,17 @@ public abstract class CassandraStore implements BackendStore {
super(provider, keyspace, store);
registerTableManager(HugeType.VERTEX,
new CassandraTables.Vertex());
new CassandraTables.Vertex(store));
registerTableManager(HugeType.EDGE_OUT,
new CassandraTables.Edge(Directions.OUT));
CassandraTables.Edge.out(store));
registerTableManager(HugeType.EDGE_IN,
new CassandraTables.Edge(Directions.IN));
CassandraTables.Edge.in(store));
registerTableManager(HugeType.SECONDARY_INDEX,
new CassandraTables.SecondaryIndex());
new CassandraTables.SecondaryIndex(store));
registerTableManager(HugeType.RANGE_INDEX,
new CassandraTables.RangeIndex());
new CassandraTables.RangeIndex(store));
}
@Override

2
hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java
View File

@ -557,7 +557,7 @@ public abstract class CassandraTable
protected void createIndex(CassandraSessionPool.Session session,
String indexLabel, HugeKeys column) {
String indexName = this.table() + "_" + indexLabel;
String indexName = joinTableName(this.table(), indexLabel);
SchemaStatement index = SchemaBuilder.createIndex(indexName)
.ifNotExists()
.onTable(this.table())

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

@ -48,6 +48,9 @@ import com.google.common.collect.ImmutableMap;
public class CassandraTables {
public static final String LABEL_INDEX = "label_index";
public static final String NAME_INDEX = "name_index";
private static final DataType DATATYPE_PK = DataType.cint();
private static final DataType DATATYPE_SL = DataType.cint(); // VL/EL
private static final DataType DATATYPE_IL = DataType.cint();
@ -142,7 +145,7 @@ public class CassandraTables {
.build();
this.createTable(session, pkeys, ckeys, columns);
this.createIndex(session, "vertex_label_name_index", HugeKeys.NAME);
this.createIndex(session, NAME_INDEX, HugeKeys.NAME);
}
}
@ -175,7 +178,7 @@ public class CassandraTables {
.build();
this.createTable(session, pkeys, ckeys, columns);
this.createIndex(session, "edge_label_name_index", HugeKeys.NAME);
this.createIndex(session, NAME_INDEX, HugeKeys.NAME);
}
}
@ -202,7 +205,7 @@ public class CassandraTables {
);
this.createTable(session, pkeys, ckeys, columns);
this.createIndex(session, "property_key_name_index", HugeKeys.NAME);
this.createIndex(session, NAME_INDEX, HugeKeys.NAME);
}
}
@ -230,7 +233,7 @@ public class CassandraTables {
);
this.createTable(session, pkeys, ckeys, columns);
this.createIndex(session, "index_label_name_index", HugeKeys.NAME);
this.createIndex(session, NAME_INDEX, HugeKeys.NAME);
}
}
@ -238,8 +241,8 @@ public class CassandraTables {
public static final String TABLE = "vertices";
public Vertex() {
super(TABLE);
public Vertex(String store) {
super(joinTableName(store, TABLE));
}
@Override
@ -255,7 +258,7 @@ public class CassandraTables {
);
this.createTable(session, pkeys, ckeys, columns);
this.createIndex(session, "vertex_label_index", HugeKeys.LABEL);
this.createIndex(session, LABEL_INDEX, HugeKeys.LABEL);
}
}
@ -263,13 +266,23 @@ public class CassandraTables {
public static final String TABLE_PREFIX = "edges";
private final String store;
private final Directions direction;
public Edge(Directions direction) {
super(table(direction));
protected Edge(String store, Directions direction) {
super(joinTableName(store, table(direction)));
this.store = store;
this.direction = direction;
}
protected String edgesTable(Directions direction) {
return joinTableName(this.store, table(direction));
}
protected Directions direction() {
return this.direction;
}
@Override
public void init(CassandraSessionPool.Session session) {
ImmutableMap<HugeKeys, DataType> pkeys = ImmutableMap.of(
@ -293,7 +306,7 @@ public class CassandraTables {
* by label from out-edges table
*/
if (this.direction == Directions.OUT) {
this.createIndex(session, "edge_label_index", HugeKeys.LABEL);
this.createIndex(session, LABEL_INDEX, HugeKeys.LABEL);
}
}
@ -395,9 +408,9 @@ public class CassandraTables {
}
}
private static Delete buildDelete(Id label, String ownerVertex,
Directions direction) {
Delete delete = QueryBuilder.delete().from(table(direction));
private Delete buildDelete(Id label, String ownerVertex,
Directions direction) {
Delete delete = QueryBuilder.delete().from(edgesTable(direction));
delete.where(formatEQ(HugeKeys.OWNER_VERTEX, ownerVertex));
delete.where(formatEQ(HugeKeys.DIRECTION, direction.code()));
delete.where(formatEQ(HugeKeys.LABEL, label.asLong()));
@ -444,18 +457,26 @@ public class CassandraTables {
return vertex;
}
public static String table(Directions direction) {
private static String table(Directions direction) {
assert direction == Directions.OUT || direction == Directions.IN;
return TABLE_PREFIX + "_" + direction.string();
}
public static CassandraTable out(String store) {
return new Edge(store, Directions.OUT);
}
public static CassandraTable in(String store) {
return new Edge(store, Directions.IN);
}
}
public static class SecondaryIndex extends CassandraTable {
public static final String TABLE = "secondary_indexes";
public SecondaryIndex() {
super(TABLE);
public SecondaryIndex(String store) {
super(joinTableName(store, TABLE));
}
@Override
@ -548,8 +569,8 @@ public class CassandraTables {
public static final String TABLE = "range_indexes";
public RangeIndex() {
super(TABLE);
public RangeIndex(String store) {
super(joinTableName(store, TABLE));
}
@Override

4
hugegraph-core/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<groupId>com.baidu.hugegraph</groupId>
<artifactId>hugegraph</artifactId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>hugegraph-core</artifactId>
@ -106,7 +106,7 @@
</manifest>
<manifestEntries>
<Implementation-Version>
0.7.0.0
0.7.1.0
</Implementation-Version>
</manifestEntries>
</archive>

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

@ -59,6 +59,7 @@ import com.baidu.hugegraph.schema.SchemaElement;
import com.baidu.hugegraph.schema.SchemaManager;
import com.baidu.hugegraph.schema.VertexLabel;
import com.baidu.hugegraph.structure.HugeFeatures;
import com.baidu.hugegraph.task.HugeTaskManager;
import com.baidu.hugegraph.traversal.optimize.HugeGraphStepStrategy;
import com.baidu.hugegraph.traversal.optimize.HugeVertexStepStrategy;
import com.baidu.hugegraph.util.E;
@ -99,6 +100,7 @@ public class HugeGraph implements Graph {
private final EventHub schemaEventHub;
private final EventHub indexEventHub;
private final RateLimiter rateLimiter;
private final HugeTaskManager taskManager;
private final HugeFeatures features;
@ -116,6 +118,8 @@ public class HugeGraph implements Graph {
final int limit = configuration.get(CoreOptions.RATE_LIMIT);
this.rateLimiter = limit > 0 ? RateLimiter.create(limit) : null;
this.taskManager = HugeTaskManager.instance();
this.features = new HugeFeatures(this, true);
this.name = configuration.get(CoreOptions.STORE);
@ -132,14 +136,9 @@ public class HugeGraph implements Graph {
this.tx = new TinkerpopTransaction(this);
this.variables = null;
}
this.taskManager.addScheduler(this);
private BackendStoreProvider loadStoreProvider() {
String backend = this.configuration.get(CoreOptions.BACKEND);
LOG.info("Opening backend store '{}' for graph '{}'",
backend, this.name);
return BackendProviderFactory.open(backend, this.name);
this.variables = null;
}
public String name() {
@ -175,28 +174,34 @@ public class HugeGraph implements Graph {
}
public void initBackend() {
this.tx.readWrite();
this.loadSchemaStore().open(this.configuration);
this.loadSystemStore().open(this.configuration);
this.loadGraphStore().open(this.configuration);
try {
this.storeProvider.init();
} finally {
this.tx.close();
this.loadGraphStore().close();
this.loadSystemStore().close();
this.loadSchemaStore().close();
}
}
public void clearBackend() {
this.tx.readWrite();
this.loadSchemaStore().open(this.configuration);
this.loadSystemStore().open(this.configuration);
this.loadGraphStore().open(this.configuration);
try {
this.storeProvider.clear();
} finally {
this.tx.close();
this.loadGraphStore().close();
this.loadSystemStore().close();
this.loadSchemaStore().close();
}
}
private SchemaTransaction openSchemaTransaction() throws HugeException {
try {
String name = this.configuration.get(CoreOptions.STORE_SCHEMA);
BackendStore store = this.storeProvider.loadSchemaStore(name);
return new CachedSchemaTransaction(this, store);
return new CachedSchemaTransaction(this, this.loadSchemaStore());
} catch (BackendException e) {
String message = "Failed to open schema transaction";
LOG.error("{}", message, e);
@ -206,9 +211,7 @@ public class HugeGraph implements Graph {
private GraphTransaction openGraphTransaction() throws HugeException {
try {
String graph = this.configuration.get(CoreOptions.STORE_GRAPH);
BackendStore store = this.storeProvider.loadGraphStore(graph);
return new CachedGraphTransaction(this, store);
return new CachedGraphTransaction(this, this.loadGraphStore());
} catch (BackendException e) {
String message = "Failed to open graph transaction";
LOG.error("{}", message, e);
@ -216,6 +219,28 @@ public class HugeGraph implements Graph {
}
}
private BackendStoreProvider loadStoreProvider() {
String backend = this.configuration.get(CoreOptions.BACKEND);
LOG.info("Opening backend store '{}' for graph '{}'",
backend, this.name);
return BackendProviderFactory.open(backend, this.name);
}
private BackendStore loadSchemaStore() {
String name = this.configuration.get(CoreOptions.STORE_SCHEMA);
return this.storeProvider.loadSchemaStore(name);
}
private BackendStore loadGraphStore() {
String graph = this.configuration.get(CoreOptions.STORE_GRAPH);
return this.storeProvider.loadGraphStore(graph);
}
public BackendStore loadSystemStore() {
String name = this.configuration.get(CoreOptions.STORE_SYSTEM);
return this.storeProvider.loadSystemStore(name);
}
public SchemaTransaction schemaTransaction() {
/*
* NOTE: each schema operation will be auto committed,
@ -238,6 +263,7 @@ public class HugeGraph implements Graph {
}
public GraphTransaction openTransaction() {
// Open a new one
return this.openGraphTransaction();
}
@ -362,6 +388,7 @@ public class HugeGraph implements Graph {
this.closeTx();
} finally {
this.storeProvider.close();
this.taskManager.closeScheduler(this);
}
}
@ -470,6 +497,7 @@ public class HugeGraph implements Graph {
*/
public static void shutdown(long timout) throws InterruptedException {
EventHub.destroy(timout);
HugeTaskManager.instance().shutdown(timout);
}
private class TinkerpopTransaction extends AbstractThreadLocalTransaction {

4
hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CachedGraphTransaction.java vendored
View File

@ -111,7 +111,7 @@ public class CachedGraphTransaction extends GraphTransaction {
}
@Override
public Vertex addVertex(HugeVertex vertex) {
public HugeVertex addVertex(HugeVertex vertex) {
// Update vertex cache
this.verticesCache.invalidate(vertex.id());
@ -146,7 +146,7 @@ public class CachedGraphTransaction extends GraphTransaction {
}
@Override
public Edge addEdge(HugeEdge edge) {
public HugeEdge addEdge(HugeEdge edge) {
// TODO: Use a more precise strategy to update the edge cache
this.edgesCache.clear();

5
hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/AbstractBackendStoreProvider.java
View File

@ -128,4 +128,9 @@ public abstract class AbstractBackendStoreProvider
E.checkNotNull(store, "store");
return store;
}
@Override
public BackendStore loadSystemStore(String name) {
return this.loadGraphStore(name);
}
}

13
hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java
View File

@ -30,10 +30,12 @@ public abstract class BackendSessionPool {
private static final Logger LOG = Log.logger(BackendSessionPool.class);
private ThreadLocal<BackendSession> threadLocalSession;
private AtomicInteger sessionCount;
private final String name;
private final ThreadLocal<BackendSession> threadLocalSession;
private final AtomicInteger sessionCount;
public BackendSessionPool() {
public BackendSessionPool(String name) {
this.name = name;
this.threadLocalSession = new ThreadLocal<>();
this.sessionCount = new AtomicInteger(0);
}
@ -102,9 +104,8 @@ public abstract class BackendSessionPool {
@Override
public String toString() {
return String.format("%s@%08X",
this.getClass().getSimpleName(),
this.hashCode());
return String.format("%s-%s@%08X", this.name,
this.getClass().getSimpleName(), this.hashCode());
}
public abstract void open(HugeConfig config) throws Exception;

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

@ -29,6 +29,8 @@ public interface BackendStoreProvider {
// Graph name (that's database name)
public String graph();
public BackendStore loadSystemStore(String name);
public BackendStore loadSchemaStore(String name);
public BackendStore loadGraphStore(String name);

4
hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendTable.java
View File

@ -106,6 +106,10 @@ public abstract class BackendTable<Session extends BackendSession, Entry> {
return type;
}
public static final String joinTableName(String prefix, String table) {
return prefix + "_" + table;
}
/*************************** abstract methods ***************************/
public abstract void init(Session session);

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

@ -308,8 +308,13 @@ public class GraphTransaction extends IndexableTransaction {
return rs;
}
@Watched("graph.addVertex-with-instance")
public Vertex addVertex(HugeVertex vertex) {
@Watched(prefix = "graph")
public HugeVertex addVertex(Object... keyValues) {
return this.addVertex(this.constructVertex(true, keyValues));
}
@Watched("graph.addVertex-instance")
public HugeVertex addVertex(HugeVertex vertex) {
this.checkOwnerThread();
// Override vertexes in local `removedVertexes`
@ -335,10 +340,11 @@ public class GraphTransaction extends IndexableTransaction {
}
@Watched(prefix = "graph")
public Vertex addVertex(Object... keyValues) {
public HugeVertex constructVertex(boolean verifyVL, Object... keyValues) {
HugeElement.ElementKeys elemKeys = HugeElement.classifyKeys(keyValues);
VertexLabel vertexLabel = this.checkVertexLabel(elemKeys.label());
VertexLabel vertexLabel = this.checkVertexLabel(elemKeys.label(),
verifyVL);
Id id = HugeVertex.getIdValue(elemKeys.id());
List<Id> keys = this.graph().mapPkName2Id(elemKeys.keys());
@ -363,7 +369,7 @@ public class GraphTransaction extends IndexableTransaction {
vertex.assignId(id);
}
return this.addVertex(vertex);
return vertex;
}
@Watched(prefix = "graph")
@ -469,7 +475,7 @@ public class GraphTransaction extends IndexableTransaction {
}
@Watched(prefix = "graph")
public Edge addEdge(HugeEdge edge) {
public HugeEdge addEdge(HugeEdge edge) {
this.checkOwnerThread();
// Override edges in local `removedEdges`
@ -957,7 +963,7 @@ public class GraphTransaction extends IndexableTransaction {
}
}
private VertexLabel checkVertexLabel(Object label) {
private VertexLabel checkVertexLabel(Object label, boolean verifyLabel) {
HugeVertexFeatures features = graph().features().vertex();
// Check Vertex label
@ -974,7 +980,9 @@ public class GraphTransaction extends IndexableTransaction {
"as the vertex label argument, but got: '%s'", label);
// The label must be an instance of String or VertexLabel
if (label instanceof String) {
ElementHelper.validateLabel((String) label);
if (verifyLabel) {
ElementHelper.validateLabel((String) label);
}
label = graph().vertexLabel((String) label);
}

8
hugegraph-core/src/main/java/com/baidu/hugegraph/config/CoreOptions.java
View File

@ -63,6 +63,14 @@ public class CoreOptions extends OptionHolder {
"hugegraph"
);
public static final ConfigOption<String> STORE_SYSTEM =
new ConfigOption<>(
"store.system",
"The system table name, which store system data.",
disallowEmpty(),
"system"
);
public static final ConfigOption<String> STORE_SCHEMA =
new ConfigOption<>(
"store.schema",

2
hugegraph-core/src/main/java/com/baidu/hugegraph/structure/HugeVertex.java
View File

@ -230,7 +230,7 @@ public class HugeVertex extends HugeElement implements Vertex, Cloneable {
*/
@Watched(prefix = "vertex")
@Override
public Edge addEdge(String label, Vertex vertex, Object... keyValues) {
public HugeEdge addEdge(String label, Vertex vertex, Object... keyValues) {
ElementKeys elemKeys = HugeElement.classifyKeys(keyValues);
// Check id (must be null)
if (elemKeys.id() != null) {

361
hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTask.java Normal file
View File

@ -0,0 +1,361 @@
/*
* Copyright 2017 HugeGraph Authors
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with this
* work for additional information regarding copyright ownership. The ASF
* licenses this file to You under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package com.baidu.hugegraph.task;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.FutureTask;
import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
import org.apache.tinkerpop.gremlin.structure.T;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.type.define.SerialEnum;
import com.baidu.hugegraph.util.E;
public class HugeTask<V> extends FutureTask<V> {
private final HugeTaskCallable<V> callable;
private String type;
private String name;
private final Id id;
private final Id parent;
private List<Id> children;
private String description;
private HugeTaskStatus status;
private int progress;
private Date create;
private Date update;
private int retries;
private String result;
public HugeTask(Id id, Id parent, HugeTaskCallable<V> callable) {
super(callable);
E.checkArgumentNotNull(id, "Task id can't be null");
E.checkArgument(id.number(), "Invalid task id type, it must be number");
this.callable = callable;
this.type = null;
this.name = null;
this.id = id;
this.parent = parent;
this.children = null;
this.description = null;
this.status = HugeTaskStatus.NEW;
this.progress = 0;
this.create = new Date();
this.update = null;
this.retries = 0;
this.result = null;
}
public Id id() {
return this.id;
}
public Id parent() {
return this.parent;
}
public List<Id> children() {
return Collections.unmodifiableList(this.children);
}
public void child(Id id) {
if (this.children == null) {
this.children = new ArrayList<>();
}
this.children.add(id);
}
public void type(String type) {
this.type = type;
}
public String type() {
return this.type;
}
public void name(String name) {
this.name = name;
}
public String name() {
return this.name;
}
public void description(String description) {
this.description = description;
}
public String description() {
return this.description;
}
public void progress(int progress) {
this.progress = progress;
}
public int progress() {
return this.progress;
}
public void createTime(Date create) {
this.create = create;
}
public Date createTime() {
return this.create;
}
public void updateTime(Date update) {
this.update = update;
}
public Date updateTime() {
return this.update;
}
public void retry() {
++this.retries;
}
public int retries() {
return this.retries;
}
@Override
public String toString() {
return String.format("HugeTask(%s)%s", this.id, this.asMap());
}
@Override
public void run() {
this.status(HugeTaskStatus.RUNNING);
super.run();
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
this.status(HugeTaskStatus.CANCELLED);
return super.cancel(mayInterruptIfRunning);
}
@Override
protected void set(V v) {
this.status(HugeTaskStatus.SUCCESS);
this.result = v.toString();
super.set(v);
}
@Override
protected void setException(Throwable t) {
this.status(HugeTaskStatus.FAILED);
this.result = t.toString();
super.setException(t);
}
protected synchronized void status(HugeTaskStatus status) {
this.status = status;
}
protected HugeTaskCallable<V> callable() {
return this.callable;
}
protected void property(String key, Object value) {
E.checkNotNull(key, "property key");
switch (key) {
case P.TYPE:
this.type = (String) value;
break;
case P.NAME:
this.name = (String) value;
break;
case P.DESCRIPTION:
this.description = (String) value;
break;
case P.STATUS:
this.status = SerialEnum.fromCode(HugeTaskStatus.class,
(byte) value);
break;
case P.PROGRESS:
this.progress = (int) value;
break;
case P.CREATE:
this.create = (Date) value;
break;
case P.UPDATE:
this.update = (Date) value;
break;
case P.RETRIES:
this.retries = (int) value;
break;
case P.RESULT:
this.result = (String) value;
break;
case P.CALLABLE:
// pass
break;
default:
throw new AssertionError("Unsupported key: " + key);
}
}
protected Object[] asArray() {
E.checkState(this.type != null, "Task type can't be null");
E.checkState(this.name != null, "Task name can't be null");
List<Object> list = new ArrayList<>(24);
list.add(T.label);
list.add(P.TASK);
list.add(T.id);
list.add(this.id);
list.add(P.TYPE);
list.add(this.type);
list.add(P.NAME);
list.add(this.name);
list.add(P.CALLABLE);
list.add(this.callable.getClass().getName());
list.add(P.STATUS);
list.add(this.status.code());
list.add(P.PROGRESS);
list.add(this.progress);
list.add(P.CREATE);
list.add(this.create);
list.add(P.RETRIES);
list.add( this.retries);
if (this.description != null) {
list.add(P.DESCRIPTION);
list.add(this.description);
}
if (this.update != null) {
list.add(P.UPDATE);
list.add(this.update);
}
if (this.result != null) {
list.add(P.RESULT);
list.add(this.result);
}
return list.toArray();
}
protected Map<String, Object> asMap() {
E.checkState(this.type != null, "Task type can't be null");
E.checkState(this.name != null, "Task name can't be null");
Map<String, Object> map = new HashMap<>();
map.put(P.LABEL, P.TASK);
map.put(P.ID, this.id);
map.put(P.TYPE, this.type);
map.put(P.NAME, this.name);
map.put(P.CALLABLE, this.callable.getClass().getName());
map.put(P.STATUS, this.status.code());
map.put(P.PROGRESS, this.progress);
map.put(P.CREATE, this.create);
map.put(P.RETRIES, this.retries);
if (this.description != null) {
map.put(P.DESCRIPTION, this.description);
}
if (this.update != null) {
map.put(P.UPDATE, this.update);
}
if (this.result != null) {
map.put(P.RESULT, this.result);
}
return map;
}
public static <V> HugeTask<V> fromVertex(Vertex vertex) {
String callableName = vertex.value(P.CALLABLE);
HugeTaskCallable<V> callable;
try {
callable = HugeTaskCallable.fromClass(callableName);
} catch (Exception e) {
callable = HugeTaskCallable.empty(e);
}
HugeTask<V> task = new HugeTask<>((Id) vertex.id(), null, callable);
for (Iterator<VertexProperty<Object>> itor = vertex.properties();
itor.hasNext();) {
VertexProperty<Object> prop = itor.next();
task.property(prop.key(), prop.value());
}
return task;
}
public static final class P {
public static final String TASK = Hidden.hide("task");
public static final String ID = T.id.getAccessor();
public static final String LABEL = T.label.getAccessor();
public static final String TYPE = "~task_type";
public static final String NAME = "~task_name";
public static final String CALLABLE = "~task_callable";
public static final String DESCRIPTION = "~task_description";
public static final String STATUS = "~task_status";
public static final String PROGRESS = "~task_progress";
public static final String CREATE = "~task_create";
public static final String UPDATE = "~task_update";
public static final String RETRIES = "~task_retries";
public static final String RESULT = "~task_result";
//public static final String PARENT = hide("parent");
//public static final String CHILDREN = hide("children");
public static String hide(String key) {
return Hidden.hide("task" + "_" + key);
}
public static String unhide(String key) {
final String prefix = Hidden.hide("task" + "_");
if (key.startsWith(prefix)) {
return key.substring(prefix.length());
}
return key;
}
}
}

65
hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskCallable.java Normal file
View File

@ -0,0 +1,65 @@
/*
* Copyright 2017 HugeGraph Authors
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with this
* work for additional information regarding copyright ownership. The ASF
* licenses this file to You under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package com.baidu.hugegraph.task;
import java.util.concurrent.Callable;
public abstract class HugeTaskCallable<V> implements Callable<V> {
private HugeTaskScheduler scheduler = null;
private HugeTask<V> task = null;
public HugeTaskCallable() {
// pass
}
protected void scheduler(HugeTaskScheduler scheduler) {
this.scheduler = scheduler;
}
public HugeTaskScheduler scheduler() {
return this.scheduler;
}
protected void task(HugeTask<V> task) {
this.task = task;
}
public HugeTask<V> task() {
return this.task;
}
@SuppressWarnings("unchecked")
public static <V> HugeTaskCallable<V> fromClass(String className) throws
ClassNotFoundException, InstantiationException, IllegalAccessException {
Class<?> clazz = Class.forName(className);
return (HugeTaskCallable<V>) clazz.newInstance();
}
public static <V> HugeTaskCallable<V> empty(Exception e) {
return new HugeTaskCallable<V>() {
@Override
public V call() throws Exception {
throw e;
}
};
}
}

94
hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskManager.java Normal file
View File

@ -0,0 +1,94 @@
/*
* Copyright 2017 HugeGraph Authors
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with this
* work for additional information regarding copyright ownership. The ASF
* licenses this file to You under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package com.baidu.hugegraph.task;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import com.baidu.hugegraph.HugeException;
import com.baidu.hugegraph.HugeGraph;
public class HugeTaskManager {
private static final HugeTaskManager INSTANCE = new HugeTaskManager(4);
private final Map<HugeGraph, HugeTaskScheduler> schedulers;
private final ExecutorService taskExecutor;
private final ExecutorService dbExecutor;
public static HugeTaskManager instance() {
return INSTANCE;
}
private HugeTaskManager(int pool) {
this.schedulers = new HashMap<>();
this.taskExecutor = Executors.newFixedThreadPool(pool);
this.dbExecutor = Executors.newFixedThreadPool(1);
}
public void addScheduler(HugeGraph graph) {
ExecutorService task = this.taskExecutor;
ExecutorService db = this.dbExecutor;
this.schedulers.put(graph, new HugeTaskScheduler(graph, task, db));
}
public void closeScheduler(HugeGraph graph) {
HugeTaskScheduler scheduler = this.schedulers.get(graph);
if (scheduler != null && scheduler.close()) {
this.schedulers.remove(graph);
}
}
public HugeTaskScheduler getScheduler(HugeGraph graph) {
return this.schedulers.get(graph);
}
public void shutdown(long timeout) {
Throwable ex = null;
assert this.schedulers.isEmpty() : this.schedulers.size();
if (!this.taskExecutor.isShutdown()) {
this.taskExecutor.shutdown();
try {
this.taskExecutor.awaitTermination(timeout, TimeUnit.SECONDS);
} catch (Throwable e) {
ex = e;
}
}
if (!this.dbExecutor.isShutdown()) {
this.dbExecutor.shutdown();
try {
this.dbExecutor.awaitTermination(timeout, TimeUnit.SECONDS);
} catch (Throwable e) {
ex = e;
}
}
if (ex != null) {
throw new HugeException("Failed to wait for TaskScheduler", ex);
}
}
}

305
hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskScheduler.java Normal file
View File

@ -0,0 +1,305 @@
/*
* Copyright 2017 HugeGraph Authors
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with this
* work for additional information regarding copyright ownership. The ASF
* licenses this file to You under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package com.baidu.hugegraph.task;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import com.baidu.hugegraph.HugeException;
import com.baidu.hugegraph.HugeGraph;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.backend.query.Condition;
import com.baidu.hugegraph.backend.query.ConditionQuery;
import com.baidu.hugegraph.backend.store.BackendStore;
import com.baidu.hugegraph.backend.tx.GraphTransaction;
import com.baidu.hugegraph.iterator.MapperIterator;
import com.baidu.hugegraph.schema.IndexLabel;
import com.baidu.hugegraph.schema.PropertyKey;
import com.baidu.hugegraph.schema.SchemaManager;
import com.baidu.hugegraph.schema.VertexLabel;
import com.baidu.hugegraph.structure.HugeVertex;
import com.baidu.hugegraph.task.HugeTask.P;
import com.baidu.hugegraph.type.HugeType;
import com.baidu.hugegraph.type.define.Cardinality;
import com.baidu.hugegraph.type.define.DataType;
import com.baidu.hugegraph.type.define.HugeKeys;
import com.baidu.hugegraph.util.E;
import com.baidu.hugegraph.util.Events;
public class HugeTaskScheduler {
private final HugeGraph graph;
private final ExecutorService taskExecutor;
private final ExecutorService dbExecutor;
private final Map<Id, HugeTask<?>> tasks;
private TaskTransaction taskTx;
public HugeTaskScheduler(HugeGraph graph,
ExecutorService taskExecutor,
ExecutorService dbExecutor) {
this.graph = graph;
this.taskExecutor = taskExecutor;
this.dbExecutor = dbExecutor;
this.tasks = new HashMap<>();
this.taskTx = null;
this.listenChanges();
}
private TaskTransaction tx() {
// NOTE: only the owner thread can access task tx
if (this.taskTx == null) {
synchronized (this) {
if (this.taskTx == null) {
BackendStore store = this.graph.loadSystemStore();
this.taskTx = new TaskTransaction(this.graph, store);
}
}
}
assert this.taskTx != null;
return this.taskTx;
}
private void listenChanges() {
// Listen store event: "store.init"
this.graph.loadSystemStore().provider().listen(event -> {
if (Events.STORE_INIT.equals(event.name())) {
this.submit(() -> this.tx().initSchema());
return true;
}
return false;
});
}
public <V> Future<?> restore(HugeTask<V> task) {
E.checkArgumentNotNull(task, "Task can't be null");
task.status(HugeTaskStatus.RESTORING);
return this.submit(task);
}
public <V> Future<?> schedule(HugeTask<V> task) {
E.checkArgumentNotNull(task, "Task can't be null");
task.status(HugeTaskStatus.QUEUED);
return this.submitTask(task);
}
private <V> Future<?> submitTask(HugeTask<V> task) {
this.tasks.put(task.id(), task);
task.callable().scheduler(this);
task.callable().task(task);
return this.taskExecutor.submit(task);
}
public <V> void cancel(HugeTask<V> task) {
E.checkArgumentNotNull(task, "Task can't be null");
this.tasks.remove(task.id());
task.cancel(false);
}
public <V> void save(HugeTask<V> task) {
E.checkArgumentNotNull(task, "Task can't be null");
this.submit(() -> {
// Construct vertex from task
HugeVertex vertex = this.tx().constructVertex(task);
// Delete the old record if exist
Iterator<Vertex> old = this.tx().queryVertices(vertex.id());
if (old.hasNext()) {
HugeVertex oldV = (HugeVertex) old.next();
assert !old.hasNext();
if (this.tx().indexValueChanged(oldV, vertex)) {
// Only delete vertex if index value changed else override
this.tx().removeVertex(oldV);
}
}
// Do update
return this.tx().addVertex(vertex);
});
}
public boolean close() {
if (!this.dbExecutor.isShutdown()) {
this.submit(() -> {
this.tx().close();
this.graph.closeTx();
});
}
return true;
}
public <V> HugeTask<V> task(Id id) {
@SuppressWarnings("unchecked")
HugeTask<V> task = (HugeTask<V>) this.tasks.get(id);
if (task != null) {
return task;
}
return this.findTask(id);
}
public <V> HugeTask<V> findTask(Id id) {
return this.submit(() -> {
HugeTask<V> task = null;
Iterator<Vertex> vertices = this.tx().queryVertices(id);
if (vertices.hasNext()) {
task = HugeTask.fromVertex(vertices.next());
assert !vertices.hasNext();
}
return task;
});
}
public <V> Iterator<HugeTask<V>> findTask(HugeTaskStatus status) {
return this.queryTask(P.STATUS, status.code());
}
private <V> Iterator<HugeTask<V>> queryTask(String key, Object value) {
return this.submit(() -> {
VertexLabel vl = this.graph.vertexLabel(TaskTransaction.TASK);
PropertyKey pk = this.graph.propertyKey(key);
ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
query.showHidden(true);
query.eq(HugeKeys.LABEL, vl.id());
query.query(Condition.eq(pk.id(), value));
Iterator<Vertex> vertices = this.tx().queryVertices(query);
return new MapperIterator<>(vertices, v -> {
return HugeTask.fromVertex(v);
});
});
}
private <V> V submit(Runnable runnable) {
return this.submit(Executors.callable(runnable, null));
}
private <V> V submit(Callable<V> callable) {
try {
return this.dbExecutor.submit(callable).get();
} catch (Exception e) {
throw new HugeException("Failed to update/query TaskStore", e);
}
}
private static class TaskTransaction extends GraphTransaction {
public static final String TASK = P.TASK;
public TaskTransaction(HugeGraph graph, BackendStore store) {
super(graph, store);
this.autoCommit(true);
}
public HugeVertex constructVertex(HugeTask<?> task) {
return this.constructVertex(false, task.asArray());
}
public boolean indexValueChanged(Vertex oldV, HugeVertex newV) {
if (!oldV.value(P.STATUS).equals(newV.value(P.STATUS))) {
return true;
}
return false;
}
protected void initSchema() {
HugeGraph graph = this.graph();
VertexLabel label = graph.schemaTransaction().getVertexLabel(TASK);
if (label != null) {
return;
}
SchemaManager schema = graph.schema();
String[] properties = this.initProperties();
// Create vertex label '~task'
label = schema.vertexLabel(TASK)
.properties(properties)
.useCustomizeNumberId()
.nullableKeys(P.DESCRIPTION, P.UPDATE, P.RESULT)
.enableLabelIndex(false)
.build();
graph.schemaTransaction().addVertexLabel(label);
// Create index
this.createIndex(label, P.STATUS);
}
private String[] initProperties() {
List<String> props = new ArrayList<>();
props.add(createPropertyKey(P.TYPE));
props.add(createPropertyKey(P.NAME));
props.add(createPropertyKey(P.CALLABLE));
props.add(createPropertyKey(P.DESCRIPTION));
props.add(createPropertyKey(P.STATUS, DataType.BYTE));
props.add(createPropertyKey(P.PROGRESS, DataType.INT));
props.add(createPropertyKey(P.CREATE, DataType.DATE));
props.add(createPropertyKey(P.UPDATE, DataType.DATE));
props.add(createPropertyKey(P.RETRIES, DataType.INT));
props.add(createPropertyKey(P.RESULT));
return props.toArray(new String[0]);
}
private String createPropertyKey(String name) {
return this.createPropertyKey(name, DataType.TEXT);
}
private String createPropertyKey(String name, DataType dataType) {
return this.createPropertyKey(name, dataType, Cardinality.SINGLE);
}
private String createPropertyKey(String name, DataType dataType,
Cardinality cardinality) {
HugeGraph graph = this.graph();
SchemaManager schema = graph.schema();
PropertyKey propertyKey = schema.propertyKey(name)
.dataType(dataType)
.cardinality(cardinality)
.build();
graph.schemaTransaction().addPropertyKey(propertyKey);
return name;
}
private IndexLabel createIndex(VertexLabel label, String field) {
HugeGraph graph = this.graph();
SchemaManager schema = graph.schema();
String name = Hidden.hide("task-index-by-" + field);
IndexLabel indexLabel = schema.indexLabel(name)
.on(HugeType.VERTEX_LABEL, TASK)
.by(field)
.build();
graph.schemaTransaction().addIndexLabel(label, indexLabel);
return indexLabel;
}
}
}

57
hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskStatus.java Normal file
View File

@ -0,0 +1,57 @@
/*
* Copyright 2017 HugeGraph Authors
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with this
* work for additional information regarding copyright ownership. The ASF
* licenses this file to You under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package com.baidu.hugegraph.task;
import com.baidu.hugegraph.type.define.SerialEnum;
public enum HugeTaskStatus implements SerialEnum {
UNKNOWN(0, "UNKNOWN"),
NEW(1, "new"),
QUEUED(2, "queued"),
RESTORING(4, "restoring"),
RUNNING(4, "running"),
SUCCESS(5, "success"),
CANCELLED(6, "cancelled"),
FAILED(7, "failed");
private byte status = 0;
private String name;
static {
SerialEnum.register(HugeTaskStatus.class);
}
HugeTaskStatus(int status, String name) {
assert status < 256;
this.status = (byte) status;
this.name = name;
}
@Override
public byte code() {
return this.status;
}
public String string() {
return this.name;
}
}

14
hugegraph-core/src/main/java/com/baidu/hugegraph/variables/HugeVariables.java
View File

@ -40,6 +40,7 @@ import org.slf4j.Logger;
import com.baidu.hugegraph.HugeGraph;
import com.baidu.hugegraph.backend.query.Condition;
import com.baidu.hugegraph.backend.query.ConditionQuery;
import com.baidu.hugegraph.backend.tx.GraphTransaction;
import com.baidu.hugegraph.exception.NotFoundException;
import com.baidu.hugegraph.schema.PropertyKey;
import com.baidu.hugegraph.schema.SchemaManager;
@ -272,7 +273,7 @@ public class HugeVariables implements Graph.Variables {
return StringFactory.graphVariablesString(this);
}
private void setProperties(HugeVertex vertex, String key, Object value) {
private void setProperty(HugeVertex vertex, String key, Object value) {
String suffix;
if (value instanceof List) {
suffix = UNIFORM_LIST;
@ -321,16 +322,19 @@ public class HugeVariables implements Graph.Variables {
private void createVariableVertex(String key, Object value) {
VertexLabel vl = this.graph.vertexLabel(Hidden.hide(VARIABLES));
HugeVertex vertex = new HugeVertex(this.graph.graphTransaction(),
null, vl);
GraphTransaction tx = this.graph.graphTransaction();
HugeVertex vertex = new HugeVertex(tx, null, vl);
try {
this.setProperties(vertex, key, value);
this.setProperty(vertex, key, value);
} catch (IllegalArgumentException e) {
throw Graph.Variables.Exceptions
.dataTypeOfVariableValueNotSupported(value);
}
// AUTOMATIC id
vertex.assignId(null);
this.graph.graphTransaction().addVertex(vertex);
tx.addVertex(vertex);
}
private void removeVariableVertex(HugeVertex vertex) {

2
hugegraph-dist/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<name>hugegraph-dist</name>

2
hugegraph-example/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

99
hugegraph-example/src/main/java/com/baidu/hugegraph/example/TaskExample.java Normal file
View File

@ -0,0 +1,99 @@
/*
* Copyright 2017 HugeGraph Authors
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with this
* work for additional information regarding copyright ownership. The ASF
* licenses this file to You under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package com.baidu.hugegraph.example;
import java.util.Iterator;
import org.apache.commons.collections.IteratorUtils;
import org.slf4j.Logger;
import com.baidu.hugegraph.HugeGraph;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.backend.id.IdGenerator;
import com.baidu.hugegraph.task.HugeTask;
import com.baidu.hugegraph.task.HugeTaskCallable;
import com.baidu.hugegraph.task.HugeTaskManager;
import com.baidu.hugegraph.task.HugeTaskScheduler;
import com.baidu.hugegraph.task.HugeTaskStatus;
import com.baidu.hugegraph.util.Log;
public class TaskExample {
private static final Logger LOG = Log.logger(TaskExample.class);
public static void main(String[] args) throws InterruptedException {
LOG.info("TaskExample start!");
HugeGraph graph = ExampleUtil.loadGraph();
Id id = IdGenerator.of(8);
TestTask callable = new TestTask();
HugeTask<?> task = new HugeTask<>(id, null, callable);
task.type("type-1");
task.name("test-task");
HugeTaskScheduler scheduler = HugeTaskManager.instance()
.getScheduler(graph);
scheduler.schedule(task);
scheduler.save(task);
Iterator<HugeTask<Object>> itor;
itor = scheduler.findTask(HugeTaskStatus.RUNNING);
System.out.println(">>>> running task: " + IteratorUtils.toList(itor));
Thread.sleep(TestTask.UNIT * 33);
callable.run = false;
Thread.sleep(TestTask.UNIT * 1);
scheduler.save(task);
itor = scheduler.findTask(HugeTaskStatus.SUCCESS);
if (itor.hasNext()) {
task = itor.next();
}
System.out.println(">>>> task stoped");
Thread.sleep(TestTask.UNIT * 10);
System.out.println(">>>> restore task...");
scheduler.restore(task);
Thread.sleep(TestTask.UNIT * 80);
scheduler.save(task);
graph.close();
HugeGraph.shutdown(30L);
}
public static class TestTask extends HugeTaskCallable<Integer> {
public static final int UNIT = 100;
public volatile boolean run = true;
@Override
public Integer call() throws Exception {
for (int i = this.task().progress(); i <= 100 && this.run; i++) {
System.out.println(">>>> progress " + i);
this.task().progress(i);
this.scheduler().save(this.task());
Thread.sleep(UNIT);
}
return 18;
}
}
}

2
hugegraph-hbase/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

3
hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSessions.java
View File

@ -66,7 +66,8 @@ public class HbaseSessions extends BackendSessionPool {
private final String namespace;
private Connection hbase;
public HbaseSessions(String namespace) {
public HbaseSessions(String namespace, String store) {
super(namespace + "/" + store);
this.namespace = namespace;
}

18
hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseStore.java
View File

@ -66,7 +66,7 @@ public abstract class HbaseStore implements BackendStore {
this.provider = provider;
this.namespace = namespace;
this.store = store;
this.sessions = new HbaseSessions(namespace);
this.sessions = new HbaseSessions(namespace, store);
}
protected void registerTableManager(HugeType type, HbaseTable table) {
@ -262,7 +262,7 @@ public abstract class HbaseStore implements BackendStore {
@Override
public void beginTx() {
// TODO Auto-generated method stub
// pass
}
@Override
@ -279,7 +279,7 @@ public abstract class HbaseStore implements BackendStore {
@Override
public void rollbackTx() {
// TODO Auto-generated method stub
// pass
}
@Override
@ -318,7 +318,7 @@ public abstract class HbaseStore implements BackendStore {
new HbaseTables.IndexLabel());
registerTableManager(HugeType.SECONDARY_INDEX,
new HbaseTables.SecondaryIndex());
new HbaseTables.SecondaryIndex(store));
}
@Override
@ -342,17 +342,17 @@ public abstract class HbaseStore implements BackendStore {
super(provider, namespace, store);
registerTableManager(HugeType.VERTEX,
new HbaseTables.Vertex());
new HbaseTables.Vertex(store));
registerTableManager(HugeType.EDGE_OUT,
HbaseTables.Edge.out());
HbaseTables.Edge.out(store));
registerTableManager(HugeType.EDGE_IN,
HbaseTables.Edge.in());
HbaseTables.Edge.in(store));
registerTableManager(HugeType.SECONDARY_INDEX,
new HbaseTables.SecondaryIndex());
new HbaseTables.SecondaryIndex(store));
registerTableManager(HugeType.RANGE_INDEX,
new HbaseTables.RangeIndex());
new HbaseTables.RangeIndex(store));
}
@Override

28
hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTables.java
View File

@ -100,8 +100,8 @@ public class HbaseTables {
public static final String TABLE = "v";
public Vertex() {
super(TABLE);
public Vertex(String store) {
super(joinTableName(store, TABLE));
}
}
@ -109,17 +109,21 @@ public class HbaseTables {
public static final String TABLE_SUFFIX = "e";
public Edge(boolean out) {
public Edge(String store, boolean out) {
super(joinTableName(store, table(out)));
}
private static String table(boolean out) {
// Edge out/in table
super((out ? 'o' : 'i') + TABLE_SUFFIX);
return (out ? 'o' : 'i') + TABLE_SUFFIX;
}
public static Edge out() {
return new Edge(true);
public static Edge out(String store) {
return new Edge(store, true);
}
public static Edge in() {
return new Edge(false);
public static Edge in(String store) {
return new Edge(store, false);
}
@Override
@ -148,8 +152,8 @@ public class HbaseTables {
public static final String TABLE = "si";
public SecondaryIndex() {
super(TABLE);
public SecondaryIndex(String store) {
super(joinTableName(store, TABLE));
}
}
@ -157,8 +161,8 @@ public class HbaseTables {
public static final String TABLE = "ri";
public RangeIndex() {
super(TABLE);
public RangeIndex(String store) {
super(joinTableName(store, TABLE));
}
@Override

2
hugegraph-mysql/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

21
hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlSessions.java
View File

@ -43,13 +43,14 @@ public class MysqlSessions extends BackendSessionPool {
private static final Logger LOG = Log.logger(MysqlStore.class);
private static final Integer DROP_DB_TIMEOUT = 10000;
private static final int DROP_DB_TIMEOUT = 10000;
private HugeConfig config;
private String database;
private boolean opened;
public MysqlSessions(HugeConfig config, String database) {
public MysqlSessions(HugeConfig config, String database, String store) {
super(database + "/" + store);
this.config = config;
this.database = database;
this.opened = false;
@ -183,8 +184,8 @@ public class MysqlSessions extends BackendSessionPool {
}
}
} catch (Exception e) {
throw new BackendException("Failed to obtain mysql databases " +
"info, please ensure it is ok", e);
throw new BackendException("Failed to obtain MySQL metadata, " +
"please ensure it is ok", e);
}
return false;
}
@ -192,14 +193,14 @@ public class MysqlSessions extends BackendSessionPool {
/**
* Connect DB without specified database
*/
private Connection openWithoutDB(Integer timeout) {
private Connection openWithoutDB(int timeout) {
String jdbcUrl = this.config.get(MysqlOptions.JDBC_URL);
URIBuilder url = new URIBuilder();
url.setPath(jdbcUrl).setParameter("socketTimeout", timeout.toString());
String url = new URIBuilder().setPath(jdbcUrl)
.setParameter("socketTimeout",
String.valueOf(timeout))
.toString();
try {
return connect(url.toString());
return connect(url);
} catch (SQLException e) {
throw new BackendException("Failed to access %s, " +
"please ensure it is ok", jdbcUrl);

13
hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlStore.java
View File

@ -37,7 +37,6 @@ import com.baidu.hugegraph.backend.store.BackendStore;
import com.baidu.hugegraph.backend.store.BackendStoreProvider;
import com.baidu.hugegraph.config.HugeConfig;
import com.baidu.hugegraph.type.HugeType;
import com.baidu.hugegraph.type.define.Directions;
import com.baidu.hugegraph.util.E;
import com.baidu.hugegraph.util.Log;
@ -75,7 +74,7 @@ public abstract class MysqlStore implements BackendStore {
}
protected MysqlSessions openSessionPool(HugeConfig config) {
return new MysqlSessions(config, this.database);
return new MysqlSessions(config, this.database, this.store);
}
public Map<HugeType, MysqlTable> tables() {
@ -357,17 +356,17 @@ public abstract class MysqlStore implements BackendStore {
super(provider, database, store);
registerTableManager(HugeType.VERTEX,
new MysqlTables.Vertex());
new MysqlTables.Vertex(store));
registerTableManager(HugeType.EDGE_OUT,
new MysqlTables.Edge(Directions.OUT));
MysqlTables.Edge.out(store));
registerTableManager(HugeType.EDGE_IN,
new MysqlTables.Edge(Directions.IN));
MysqlTables.Edge.in(store));
registerTableManager(HugeType.SECONDARY_INDEX,
new MysqlTables.SecondaryIndex());
new MysqlTables.SecondaryIndex(store));
registerTableManager(HugeType.RANGE_INDEX,
new MysqlTables.RangeIndex());
new MysqlTables.RangeIndex(store));
}
@Override

45
hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTables.java
View File

@ -76,10 +76,10 @@ public class MysqlTables {
public Counters() {
super(TABLE);
this.define = new TableDefine();
this.define.column(HugeKeys.SCHEMA_TYPE, VARCHAR);
this.define.column(HugeKeys.ID, INT);
// Primary keys
this.define.keys(HugeKeys.SCHEMA_TYPE);
}
@ -141,6 +141,7 @@ public class MysqlTables {
public VertexLabel() {
super(TABLE);
this.define = new TableDefine();
this.define.column(HugeKeys.ID, DATATYPE_SL);
this.define.column(HugeKeys.NAME, VARCHAR);
@ -151,7 +152,6 @@ public class MysqlTables {
this.define.column(HugeKeys.INDEX_LABELS, SMALL_JSON);
this.define.column(HugeKeys.ENABLE_LABEL_INDEX, BOOLEAN);
this.define.column(HugeKeys.USER_DATA, LARGE_JSON);
// Primary keys
this.define.keys(HugeKeys.ID);
}
}
@ -162,6 +162,7 @@ public class MysqlTables {
public EdgeLabel() {
super(TABLE);
this.define = new TableDefine();
this.define.column(HugeKeys.ID, DATATYPE_SL);
this.define.column(HugeKeys.NAME, VARCHAR);
@ -174,7 +175,6 @@ public class MysqlTables {
this.define.column(HugeKeys.INDEX_LABELS, SMALL_JSON);
this.define.column(HugeKeys.ENABLE_LABEL_INDEX, BOOLEAN);
this.define.column(HugeKeys.USER_DATA, LARGE_JSON);
// Primary keys
this.define.keys(HugeKeys.ID);
}
}
@ -185,6 +185,7 @@ public class MysqlTables {
public PropertyKey() {
super(TABLE);
this.define = new TableDefine();
this.define.column(HugeKeys.ID, DATATYPE_PK);
this.define.column(HugeKeys.NAME, VARCHAR);
@ -192,7 +193,6 @@ public class MysqlTables {
this.define.column(HugeKeys.CARDINALITY, TINYINT);
this.define.column(HugeKeys.PROPERTIES, SMALL_JSON);
this.define.column(HugeKeys.USER_DATA, LARGE_JSON);
// Primary keys
this.define.keys(HugeKeys.ID);
}
}
@ -203,6 +203,7 @@ public class MysqlTables {
public IndexLabel() {
super(TABLE);
this.define = new TableDefine();
this.define.column(HugeKeys.ID, DATATYPE_IL);
this.define.column(HugeKeys.NAME, VARCHAR);
@ -210,7 +211,6 @@ public class MysqlTables {
this.define.column(HugeKeys.BASE_VALUE, DATATYPE_SL);
this.define.column(HugeKeys.INDEX_TYPE, TINYINT);
this.define.column(HugeKeys.FIELDS, SMALL_JSON);
// Primary keys
this.define.keys(HugeKeys.ID);
}
}
@ -219,13 +219,13 @@ public class MysqlTables {
public static final String TABLE = "vertices";
public Vertex() {
super(TABLE);
public Vertex(String store) {
super(joinTableName(store, TABLE));
this.define = new TableDefine();
this.define.column(HugeKeys.ID, VARCHAR);
this.define.column(HugeKeys.LABEL, DATATYPE_SL);
this.define.column(HugeKeys.PROPERTIES, LARGE_JSON);
// Primary keys
this.define.keys(HugeKeys.ID);
}
}
@ -237,13 +237,13 @@ public class MysqlTables {
private final Directions direction;
private final String delByLabelTemplate;
public Edge(Directions direction) {
super(table(direction));
this.direction = direction;
protected Edge(String store, Directions direction) {
super(joinTableName(store, table(direction)));
this.direction = direction;
this.delByLabelTemplate = String.format(
"DELETE FROM %s WHERE %s = ?;",
table(), formatKey(HugeKeys.LABEL));
this.table(), formatKey(HugeKeys.LABEL));
this.define = new TableDefine();
this.define.column(HugeKeys.OWNER_VERTEX, VARCHAR);
@ -252,7 +252,6 @@ public class MysqlTables {
this.define.column(HugeKeys.SORT_VALUES, VARCHAR);
this.define.column(HugeKeys.OTHER_VERTEX, VARCHAR);
this.define.column(HugeKeys.PROPERTIES, LARGE_JSON);
// Primary keys
this.define.keys(HugeKeys.OWNER_VERTEX, HugeKeys.DIRECTION,
HugeKeys.LABEL, HugeKeys.SORT_VALUES,
HugeKeys.OTHER_VERTEX);
@ -359,6 +358,14 @@ public class MysqlTables {
assert direction == Directions.OUT || direction == Directions.IN;
return TABLE_PREFIX + "_" + direction.string();
}
public static MysqlTable out(String store) {
return new Edge(store, Directions.OUT);
}
public static MysqlTable in(String store) {
return new Edge(store, Directions.IN);
}
}
public abstract static class Index extends MysqlTableTemplate {
@ -395,13 +402,13 @@ public class MysqlTables {
public static final String TABLE = "secondary_indexes";
public SecondaryIndex() {
super(TABLE);
public SecondaryIndex(String store) {
super(joinTableName(store, TABLE));
this.define = new TableDefine();
this.define.column(HugeKeys.FIELD_VALUES, VARCHAR);
this.define.column(HugeKeys.INDEX_LABEL_ID, DATATYPE_IL);
this.define.column(HugeKeys.ELEMENT_IDS, VARCHAR);
// Primary keys
this.define.keys(HugeKeys.FIELD_VALUES,
HugeKeys.INDEX_LABEL_ID,
HugeKeys.ELEMENT_IDS);
@ -419,13 +426,13 @@ public class MysqlTables {
public static final String TABLE = "range_indexes";
public RangeIndex() {
super(TABLE);
public RangeIndex(String store) {
super(joinTableName(store, TABLE));
this.define = new TableDefine();
this.define.column(HugeKeys.INDEX_LABEL_ID, DATATYPE_IL);
this.define.column(HugeKeys.FIELD_VALUES, DOUBLE);
this.define.column(HugeKeys.ELEMENT_IDS, VARCHAR);
// Primary keys
this.define.keys(HugeKeys.INDEX_LABEL_ID,
HugeKeys.FIELD_VALUES,
HugeKeys.ELEMENT_IDS);

2
hugegraph-palo/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

12
hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java
View File

@ -57,9 +57,9 @@ public class PaloSessions extends MysqlSessions {
private final Timer timer;
private final PaloLoadTask loadTask;
public PaloSessions(HugeConfig config, String database,
public PaloSessions(HugeConfig config, String database, String store,
List<String> tableDirs) {
super(config, database);
super(config, database, store);
this.counter = new AtomicInteger();
this.locks = new ConcurrentHashMap<>();
// Scan disk files and restore session information
@ -168,14 +168,14 @@ public class PaloSessions extends MysqlSessions {
private int writeBatch() {
int updated = 0;
locks.get(this.id).writeLock().lock();
PaloSessions.this.locks.get(this.id).writeLock().lock();
try {
for (String table : this.batch.keySet()) {
PaloFile file = this.getOrCreate(table);
updated += file.writeLines(this.batch.get(table));
}
} finally {
locks.get(this.id).writeLock().unlock();
PaloSessions.this.locks.get(this.id).writeLock().unlock();
}
return updated;
}
@ -307,7 +307,7 @@ public class PaloSessions extends MysqlSessions {
int sessionId = file.sessionId();
LOG.info("Ready to load one batch from file: {}", file);
// Get write lock because will delete file
Lock lock = locks.get(sessionId).writeLock();
Lock lock = PaloSessions.this.locks.get(sessionId).writeLock();
lock.lock();
try {
String table = file.table();;
@ -322,7 +322,7 @@ public class PaloSessions extends MysqlSessions {
}
private String formatLabel(String table) {
return table + "-" + DATE_FORMAT.format(new Date());
return table + "-" + this.DATE_FORMAT.format(new Date());
}
}
}

3
hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloStore.java
View File

@ -42,7 +42,8 @@ public abstract class PaloStore extends MysqlStore {
@Override
protected PaloSessions openSessionPool(HugeConfig config) {
LOG.info("Open palo session pool for {}", this);
return new PaloSessions(config, this.database(), this.tableNames());
return new PaloSessions(config, this.database(), this.store(),
this.tableNames());
}
private List<String> tableNames() {

2
hugegraph-rocksdb/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

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

@ -35,7 +35,8 @@ public abstract class RocksDBSessions extends BackendSessionPool {
private final String store;
public RocksDBSessions(String store) {
public RocksDBSessions(String database, String store) {
super(database + "/" + store);
this.store = store;
}

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

@ -62,9 +62,10 @@ public class RocksDBStdSessions extends RocksDBSessions {
private final HugeConfig conf;
private final RocksDB rocksdb;
public RocksDBStdSessions(HugeConfig config, String store)
public RocksDBStdSessions(HugeConfig config, String database, String store)
throws RocksDBException {
super(store);
super(database, store);
this.conf = config;
String dataPath = wrapPath(this.conf.get(RocksDBOptions.DATA_PATH));
@ -82,9 +83,9 @@ public class RocksDBStdSessions extends RocksDBSessions {
this.rocksdb = RocksDB.open(options, dataPath);
}
public RocksDBStdSessions(HugeConfig config, String store,
public RocksDBStdSessions(HugeConfig config, String database, String store,
List<String> cfNames) throws RocksDBException {
super(store);
super(database, store);
this.conf = config;
String dataPath = wrapPath(this.conf.get(RocksDBOptions.DATA_PATH));

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

@ -201,9 +201,10 @@ public abstract class RocksDBStore implements BackendStore {
List<String> tableNames)
throws RocksDBException {
if (tableNames == null) {
return new RocksDBStdSessions(config, this.store);
return new RocksDBStdSessions(config, this.database, this.store);
} else {
return new RocksDBStdSessions(config, this.store, tableNames);
return new RocksDBStdSessions(config, this.database, this.store,
tableNames);
}
}

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

@ -51,10 +51,10 @@ public class RocksDBSstSessions extends RocksDBSessions {
private final String dataPath;
private final Map<String, SstFileWriter> tables;
public RocksDBSstSessions(HugeConfig config, String store) {
super(store);
public RocksDBSstSessions(HugeConfig conf, String database, String store) {
super(database, store);
this.conf = config;
this.conf = conf;
this.dataPath = this.wrapPath(this.conf.get(RocksDBOptions.DATA_PATH));
this.tables = new ConcurrentHashMap<>();
@ -64,9 +64,9 @@ public class RocksDBSstSessions extends RocksDBSessions {
}
}
public RocksDBSstSessions(HugeConfig config, String store,
public RocksDBSstSessions(HugeConfig config, String database, String store,
List<String> tableNames) throws RocksDBException {
this(config, store);
this(config, database, store);
for (String table : tableNames) {
this.createTable(table);
}

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

@ -43,9 +43,11 @@ public abstract class RocksDBSstStore extends RocksDBStore {
List<String> tableNames)
throws RocksDBException {
if (tableNames == null) {
return new RocksDBSstSessions(config, this.store());
return new RocksDBSstSessions(config, this.database(),
this.store());
} else {
return new RocksDBSstSessions(config, this.store(), tableNames);
return new RocksDBSstSessions(config, this.database(),
this.store(), tableNames);
}
}

2
hugegraph-scylladb/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

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

@ -108,18 +108,18 @@ public class ScyllaDBStoreProvider extends CassandraStoreProvider {
if (version >= 20) {
registerTableManager(HugeType.VERTEX,
new ScyllaDBTablesWithMV.Vertex());
new ScyllaDBTablesWithMV.Vertex(store));
registerTableManager(HugeType.EDGE_OUT,
ScyllaDBTablesWithMV.Edge.out());
ScyllaDBTablesWithMV.Edge.out(store));
registerTableManager(HugeType.EDGE_IN,
ScyllaDBTablesWithMV.Edge.in());
ScyllaDBTablesWithMV.Edge.in(store));
} else {
registerTableManager(HugeType.EDGE_OUT,
new ScyllaDBTables.Vertex());
new ScyllaDBTables.Vertex(store));
registerTableManager(HugeType.EDGE_OUT,
ScyllaDBTables.Edge.out());
ScyllaDBTables.Edge.out(store));
registerTableManager(HugeType.EDGE_IN,
ScyllaDBTables.Edge.in());
ScyllaDBTables.Edge.in(store));
}
}

77
hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTables.java
View File

@ -166,18 +166,23 @@ public class ScyllaDBTables {
public static class Vertex extends CassandraTables.Vertex {
private static final String LIDX_TABLE = "vertex_label_index";
public Vertex(String store) {
super(store);
}
@Override
protected void createIndex(CassandraSessionPool.Session session,
String indexLabel,
HugeKeys column) {
createIndexTable(session, LIDX_TABLE);
String indexLabel, HugeKeys column) {
createIndexTable(session, this.indexTable());
}
private String indexTable() {
return joinTableName(this.table(), CassandraTables.LABEL_INDEX);
}
@Override
public void dropTable(CassandraSessionPool.Session session) {
session.execute(SchemaBuilder.dropTable(LIDX_TABLE).ifExists());
session.execute(SchemaBuilder.dropTable(indexTable()).ifExists());
super.dropTable(session);
}
@ -185,13 +190,13 @@ public class ScyllaDBTables {
public void insert(CassandraSessionPool.Session session,
CassandraBackendEntry.Row entry) {
super.insert(session, entry);
appendLabelIndex(session, LIDX_TABLE, entry);
appendLabelIndex(session, indexTable(), entry);
}
@Override
public void delete(CassandraSessionPool.Session session,
CassandraBackendEntry.Row entry) {
removeLabelIndex(session, LIDX_TABLE, entry);
removeLabelIndex(session, indexTable(), entry);
super.delete(session, entry);
}
@ -199,7 +204,7 @@ public class ScyllaDBTables {
public Iterator<BackendEntry> query(
CassandraSessionPool.Session session,
Query query) {
query = queryByLabelIndex(session, LIDX_TABLE, query);
query = queryByLabelIndex(session, indexTable(), query);
if (query == null) {
return ImmutableList.<BackendEntry>of().iterator();
}
@ -209,22 +214,24 @@ public class ScyllaDBTables {
public static class Edge extends CassandraTables.Edge {
private static final String LIDX_TABLE = "edge_label_index";
public Edge(Directions direction) {
super(direction);
public Edge(String store, Directions direction) {
super(store, direction);
}
@Override
protected void createIndex(CassandraSessionPool.Session session,
String indexLabel,
HugeKeys column) {
createIndexTable(session, LIDX_TABLE);
String indexLabel, HugeKeys column) {
assert this.direction() == Directions.OUT;
createIndexTable(session, this.indexTable());
}
private String indexTable() {
return joinTableName(this.table(), CassandraTables.LABEL_INDEX);
}
@Override
public void dropTable(CassandraSessionPool.Session session) {
session.execute(SchemaBuilder.dropTable(LIDX_TABLE).ifExists());
session.execute(SchemaBuilder.dropTable(indexTable()).ifExists());
super.dropTable(session);
}
@ -235,29 +242,35 @@ public class ScyllaDBTables {
Byte dir = entry.column(HugeKeys.DIRECTION);
Directions direction = SerialEnum.fromCode(Directions.class, dir);
if (direction == Directions.OUT) {
appendLabelIndex(session, LIDX_TABLE, entry);
appendLabelIndex(session, indexTable(), entry);
}
}
@Override
public void delete(CassandraSessionPool.Session session,
CassandraBackendEntry.Row entry) {
removeLabelIndex(session, LIDX_TABLE, entry);
if (this.direction() == Directions.OUT) {
removeLabelIndex(session, indexTable(), entry);
}
super.delete(session, entry);
}
@Override
protected void deleteEdgesByLabel(CassandraSessionPool.Session session,
Id label) {
// Edges in edges_in table will be deleted when direction is OUT
if (this.direction() == Directions.IN) {
return;
}
// Query edge id(s) by label index
Set<String> ids = queryByLabelIndex(session, LIDX_TABLE, label);
Set<String> ids = queryByLabelIndex(session, indexTable(), label);
if (ids.isEmpty()) {
return;
}
// Delete index
Delete del = QueryBuilder.delete().from(LIDX_TABLE);
Delete del = QueryBuilder.delete().from(indexTable());
del.where(formatEQ(HugeKeys.LABEL, label.asLong()));
session.add(del);
@ -269,7 +282,7 @@ public class ScyllaDBTables {
assert idNames.size() == idValues.size();
// Delete edges in OUT and IN table
String table = table(id.direction());
String table = this.edgesTable(id.direction());
Delete delete = QueryBuilder.delete().from(table);
for (int i = 0, n = idNames.size(); i < n; i++) {
delete.where(formatEQ(idNames.get(i), idValues.get(i)));
@ -289,7 +302,7 @@ public class ScyllaDBTables {
@Override
public Iterator<BackendEntry> query(
CassandraSessionPool.Session session, Query query) {
query = queryByLabelIndex(session, LIDX_TABLE, query);
query = queryByLabelIndex(session, indexTable(), query);
if (query == null) {
return ImmutableList.<BackendEntry>of().iterator();
@ -297,12 +310,12 @@ public class ScyllaDBTables {
return super.query(session, query);
}
public static Edge out() {
return new Edge(Directions.OUT);
public static Edge out(String store) {
return new Edge(store, Directions.OUT);
}
public static Edge in() {
return new Edge(Directions.IN);
public static Edge in(String store) {
return new Edge(store, Directions.IN);
}
}
@ -412,8 +425,7 @@ public class ScyllaDBTables {
@Override
protected void createIndex(CassandraSessionPool.Session session,
String indexLabel,
HugeKeys column) {
String indexLabel, HugeKeys column) {
this.schema.createIndex(session);
}
@ -454,8 +466,7 @@ public class ScyllaDBTables {
@Override
protected void createIndex(CassandraSessionPool.Session session,
String indexLabel,
HugeKeys column) {
String indexLabel, HugeKeys column) {
this.schema.createIndex(session);
}
@ -496,8 +507,7 @@ public class ScyllaDBTables {
@Override
protected void createIndex(CassandraSessionPool.Session session,
String indexLabel,
HugeKeys column) {
String indexLabel, HugeKeys column) {
this.schema.createIndex(session);
}
@ -538,8 +548,7 @@ public class ScyllaDBTables {
@Override
protected void createIndex(CassandraSessionPool.Session session,
String indexLabel,
HugeKeys column) {
String indexLabel, HugeKeys column) {
this.schema.createIndex(session);
}

16
hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTablesWithMV.java
View File

@ -55,6 +55,10 @@ public class ScyllaDBTablesWithMV {
private static final String MV_LABEL2VERTEX = "mv_label2vertex";
public Vertex(String store) {
super(store);
}
@Override
protected void createIndex(CassandraSessionPool.Session session,
String indexLabel,
@ -107,8 +111,8 @@ public class ScyllaDBTablesWithMV {
private final String PRKEYS_NN = this.KEYS.stream().collect(
Collectors.joining(" IS NOT NULL AND "));
public Edge(Directions direction) {
super(direction);
public Edge(String store, Directions direction) {
super(store, direction);
}
@Override
@ -147,12 +151,12 @@ public class ScyllaDBTablesWithMV {
return super.query2Select(table, query);
}
public static Edge out() {
return new Edge(Directions.OUT);
public static Edge out(String store) {
return new Edge(store, Directions.OUT);
}
public static Edge in() {
return new Edge(Directions.IN);
public static Edge in(String store) {
return new Edge(store, Directions.IN);
}
}
}

2
hugegraph-test/pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<artifactId>hugegraph</artifactId>
<groupId>com.baidu.hugegraph</groupId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

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

@ -119,7 +119,7 @@ public class BaseRocksDBUnitTest extends BaseUnitTest {
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, "test-store");
RocksDBSessions rocks = new RocksDBStdSessions(config, "db", "store");
rocks.createTable(table);
return rocks;
}

2
pom.xml
View File

@ -4,7 +4,7 @@
<modelVersion>4.0.0</modelVersion>
<groupId>com.baidu.hugegraph</groupId>
<artifactId>hugegraph</artifactId>
<version>0.7.0-SNAPSHOT</version>
<version>0.7.1-SNAPSHOT</version>
<packaging>pom</packaging>
<prerequisites>
<maven>3.3.9</maven>