diff --git a/hugegraph-api/pom.xml b/hugegraph-api/pom.xml index 452a3d8d0..e8891ccf7 100644 --- a/hugegraph-api/pom.xml +++ b/hugegraph-api/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 diff --git a/hugegraph-cassandra/pom.xml b/hugegraph-cassandra/pom.xml index 0a574fc6d..64da229df 100644 --- a/hugegraph-cassandra/pom.xml +++ b/hugegraph-cassandra/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 diff --git a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraSessionPool.java b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraSessionPool.java index 92e70e4a0..b80c1481a 100644 --- a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraSessionPool.java +++ b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraSessionPool.java @@ -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 diff --git a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraStore.java b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraStore.java index 9ee363102..8c2d05a57 100644 --- a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraStore.java +++ b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraStore.java @@ -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 diff --git a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java index f26bff78c..fe9f01084 100644 --- a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java +++ b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTable.java @@ -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()) diff --git a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTables.java b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTables.java index a970d1bc6..a68344821 100644 --- a/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTables.java +++ b/hugegraph-cassandra/src/main/java/com/baidu/hugegraph/backend/store/cassandra/CassandraTables.java @@ -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 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 diff --git a/hugegraph-core/pom.xml b/hugegraph-core/pom.xml index 8baba85af..b6fa1f28b 100644 --- a/hugegraph-core/pom.xml +++ b/hugegraph-core/pom.xml @@ -5,7 +5,7 @@ com.baidu.hugegraph hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT ../pom.xml hugegraph-core @@ -106,7 +106,7 @@ - 0.7.0.0 + 0.7.1.0 diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/HugeGraph.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/HugeGraph.java index 5eaffa8f6..b5d564f70 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/HugeGraph.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/HugeGraph.java @@ -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 { diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CachedGraphTransaction.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CachedGraphTransaction.java index 4fb319fd5..a611ff260 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CachedGraphTransaction.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CachedGraphTransaction.java @@ -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(); diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/AbstractBackendStoreProvider.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/AbstractBackendStoreProvider.java index 10eb7d809..e717daadd 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/AbstractBackendStoreProvider.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/AbstractBackendStoreProvider.java @@ -128,4 +128,9 @@ public abstract class AbstractBackendStoreProvider E.checkNotNull(store, "store"); return store; } + + @Override + public BackendStore loadSystemStore(String name) { + return this.loadGraphStore(name); + } } diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java index 2da4a13c2..f875bb34d 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java @@ -30,10 +30,12 @@ public abstract class BackendSessionPool { private static final Logger LOG = Log.logger(BackendSessionPool.class); - private ThreadLocal threadLocalSession; - private AtomicInteger sessionCount; + private final String name; + private final ThreadLocal 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; diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStoreProvider.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStoreProvider.java index b0fb02683..336c7104b 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStoreProvider.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStoreProvider.java @@ -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); diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendTable.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendTable.java index b7fc9d56c..00bbae730 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendTable.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendTable.java @@ -106,6 +106,10 @@ public abstract class BackendTable { return type; } + public static final String joinTableName(String prefix, String table) { + return prefix + "_" + table; + } + /*************************** abstract methods ***************************/ public abstract void init(Session session); diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphTransaction.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphTransaction.java index de0c01ce9..32ae05694 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphTransaction.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/GraphTransaction.java @@ -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 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); } diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/config/CoreOptions.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/config/CoreOptions.java index 77dcd42da..db1341678 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/config/CoreOptions.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/config/CoreOptions.java @@ -63,6 +63,14 @@ public class CoreOptions extends OptionHolder { "hugegraph" ); + public static final ConfigOption STORE_SYSTEM = + new ConfigOption<>( + "store.system", + "The system table name, which store system data.", + disallowEmpty(), + "system" + ); + public static final ConfigOption STORE_SCHEMA = new ConfigOption<>( "store.schema", diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/structure/HugeVertex.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/structure/HugeVertex.java index bf8de29c4..8c6acdde5 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/structure/HugeVertex.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/structure/HugeVertex.java @@ -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) { diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTask.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTask.java new file mode 100644 index 000000000..cea8ff957 --- /dev/null +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTask.java @@ -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 extends FutureTask { + + private final HugeTaskCallable callable; + + private String type; + private String name; + private final Id id; + private final Id parent; + private List 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 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 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 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 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 asMap() { + E.checkState(this.type != null, "Task type can't be null"); + E.checkState(this.name != null, "Task name can't be null"); + + Map 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 HugeTask fromVertex(Vertex vertex) { + String callableName = vertex.value(P.CALLABLE); + HugeTaskCallable callable; + try { + callable = HugeTaskCallable.fromClass(callableName); + } catch (Exception e) { + callable = HugeTaskCallable.empty(e); + } + + HugeTask task = new HugeTask<>((Id) vertex.id(), null, callable); + for (Iterator> itor = vertex.properties(); + itor.hasNext();) { + VertexProperty 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; + } + } +} diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskCallable.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskCallable.java new file mode 100644 index 000000000..57d8f75d0 --- /dev/null +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskCallable.java @@ -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 implements Callable { + + private HugeTaskScheduler scheduler = null; + private HugeTask task = null; + + public HugeTaskCallable() { + // pass + } + + protected void scheduler(HugeTaskScheduler scheduler) { + this.scheduler = scheduler; + } + + public HugeTaskScheduler scheduler() { + return this.scheduler; + } + + protected void task(HugeTask task) { + this.task = task; + } + + public HugeTask task() { + return this.task; + } + + @SuppressWarnings("unchecked") + public static HugeTaskCallable fromClass(String className) throws + ClassNotFoundException, InstantiationException, IllegalAccessException { + Class clazz = Class.forName(className); + return (HugeTaskCallable) clazz.newInstance(); + } + + public static HugeTaskCallable empty(Exception e) { + return new HugeTaskCallable() { + + @Override + public V call() throws Exception { + throw e; + } + }; + } +} diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskManager.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskManager.java new file mode 100644 index 000000000..0690c689d --- /dev/null +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskManager.java @@ -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 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); + } + } +} diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskScheduler.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskScheduler.java new file mode 100644 index 000000000..7ecde1703 --- /dev/null +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskScheduler.java @@ -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> 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 Future restore(HugeTask task) { + E.checkArgumentNotNull(task, "Task can't be null"); + task.status(HugeTaskStatus.RESTORING); + return this.submit(task); + } + + public Future schedule(HugeTask task) { + E.checkArgumentNotNull(task, "Task can't be null"); + task.status(HugeTaskStatus.QUEUED); + return this.submitTask(task); + } + + private Future submitTask(HugeTask task) { + this.tasks.put(task.id(), task); + task.callable().scheduler(this); + task.callable().task(task); + return this.taskExecutor.submit(task); + } + + public void cancel(HugeTask task) { + E.checkArgumentNotNull(task, "Task can't be null"); + this.tasks.remove(task.id()); + task.cancel(false); + } + + public void save(HugeTask 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 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 HugeTask task(Id id) { + @SuppressWarnings("unchecked") + HugeTask task = (HugeTask) this.tasks.get(id); + if (task != null) { + return task; + } + return this.findTask(id); + } + + public HugeTask findTask(Id id) { + return this.submit(() -> { + HugeTask task = null; + Iterator vertices = this.tx().queryVertices(id); + if (vertices.hasNext()) { + task = HugeTask.fromVertex(vertices.next()); + assert !vertices.hasNext(); + } + return task; + }); + } + + public Iterator> findTask(HugeTaskStatus status) { + return this.queryTask(P.STATUS, status.code()); + } + + private Iterator> 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 vertices = this.tx().queryVertices(query); + return new MapperIterator<>(vertices, v -> { + return HugeTask.fromVertex(v); + }); + }); + } + + private V submit(Runnable runnable) { + return this.submit(Executors.callable(runnable, null)); + } + + private V submit(Callable 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 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; + } + } +} diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskStatus.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskStatus.java new file mode 100644 index 000000000..926b570c7 --- /dev/null +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTaskStatus.java @@ -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; + } +} diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/variables/HugeVariables.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/variables/HugeVariables.java index 20ce5579b..325444282 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/variables/HugeVariables.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/variables/HugeVariables.java @@ -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) { diff --git a/hugegraph-dist/pom.xml b/hugegraph-dist/pom.xml index 93f398d21..f79ea39a1 100644 --- a/hugegraph-dist/pom.xml +++ b/hugegraph-dist/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 hugegraph-dist diff --git a/hugegraph-example/pom.xml b/hugegraph-example/pom.xml index bb425a2dd..5ef984b23 100644 --- a/hugegraph-example/pom.xml +++ b/hugegraph-example/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 diff --git a/hugegraph-example/src/main/java/com/baidu/hugegraph/example/TaskExample.java b/hugegraph-example/src/main/java/com/baidu/hugegraph/example/TaskExample.java new file mode 100644 index 000000000..67eeb4939 --- /dev/null +++ b/hugegraph-example/src/main/java/com/baidu/hugegraph/example/TaskExample.java @@ -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> 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 { + + 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; + } + } +} diff --git a/hugegraph-hbase/pom.xml b/hugegraph-hbase/pom.xml index 936d53f03..ffc1c6536 100644 --- a/hugegraph-hbase/pom.xml +++ b/hugegraph-hbase/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 diff --git a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSessions.java b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSessions.java index 33fcefd17..1cbd884cf 100644 --- a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSessions.java +++ b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseSessions.java @@ -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; } diff --git a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseStore.java b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseStore.java index 3e4784076..3c70d43ec 100644 --- a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseStore.java +++ b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseStore.java @@ -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 diff --git a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTables.java b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTables.java index bde7eb2bd..2cb2dc80a 100644 --- a/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTables.java +++ b/hugegraph-hbase/src/main/java/com/baidu/hugegraph/backend/store/hbase/HbaseTables.java @@ -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 diff --git a/hugegraph-mysql/pom.xml b/hugegraph-mysql/pom.xml index a7e5bc70a..c93dd5126 100644 --- a/hugegraph-mysql/pom.xml +++ b/hugegraph-mysql/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 diff --git a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlSessions.java b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlSessions.java index 18c96cc5d..4ba741c54 100644 --- a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlSessions.java +++ b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlSessions.java @@ -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); diff --git a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlStore.java b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlStore.java index d6cfbcde4..7ecbfc3fc 100644 --- a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlStore.java +++ b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlStore.java @@ -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 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 diff --git a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTables.java b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTables.java index 26f9b6be3..1c9348a39 100644 --- a/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTables.java +++ b/hugegraph-mysql/src/main/java/com/baidu/hugegraph/backend/store/mysql/MysqlTables.java @@ -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); diff --git a/hugegraph-palo/pom.xml b/hugegraph-palo/pom.xml index 0c6a2d0e2..844895742 100644 --- a/hugegraph-palo/pom.xml +++ b/hugegraph-palo/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 diff --git a/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java b/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java index a3a0fc5d1..bb5454c80 100644 --- a/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java +++ b/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java @@ -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 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()); } } } diff --git a/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloStore.java b/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloStore.java index 90674753a..1bee5ec50 100644 --- a/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloStore.java +++ b/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloStore.java @@ -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 tableNames() { diff --git a/hugegraph-rocksdb/pom.xml b/hugegraph-rocksdb/pom.xml index 89b67976a..0ff24b9fb 100644 --- a/hugegraph-rocksdb/pom.xml +++ b/hugegraph-rocksdb/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 diff --git a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBSessions.java b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBSessions.java index 96753bde8..caca4d76e 100644 --- a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBSessions.java +++ b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBSessions.java @@ -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; } diff --git a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStdSessions.java b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStdSessions.java index c2ad06a92..bd3f32909 100644 --- a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStdSessions.java +++ b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStdSessions.java @@ -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 cfNames) throws RocksDBException { - super(store); + super(database, store); this.conf = config; String dataPath = wrapPath(this.conf.get(RocksDBOptions.DATA_PATH)); diff --git a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java index fde8b89d6..694f6edca 100644 --- a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java +++ b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java @@ -201,9 +201,10 @@ public abstract class RocksDBStore implements BackendStore { List 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); } } diff --git a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstSessions.java b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstSessions.java index e0f5ee617..f6d3617d9 100644 --- a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstSessions.java +++ b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstSessions.java @@ -51,10 +51,10 @@ public class RocksDBSstSessions extends RocksDBSessions { private final String dataPath; private final Map 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 tableNames) throws RocksDBException { - this(config, store); + this(config, database, store); for (String table : tableNames) { this.createTable(table); } diff --git a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstStore.java b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstStore.java index 6ee603902..6ce77f040 100644 --- a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstStore.java +++ b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdbsst/RocksDBSstStore.java @@ -43,9 +43,11 @@ public abstract class RocksDBSstStore extends RocksDBStore { List 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); } } diff --git a/hugegraph-scylladb/pom.xml b/hugegraph-scylladb/pom.xml index 47c5d300e..a2577b8b7 100644 --- a/hugegraph-scylladb/pom.xml +++ b/hugegraph-scylladb/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 diff --git a/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBStoreProvider.java b/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBStoreProvider.java index 25e6bdf4e..d50f07f05 100644 --- a/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBStoreProvider.java +++ b/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBStoreProvider.java @@ -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)); } } diff --git a/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTables.java b/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTables.java index 7815eaf18..5d72ddcf6 100644 --- a/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTables.java +++ b/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTables.java @@ -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 query( CassandraSessionPool.Session session, Query query) { - query = queryByLabelIndex(session, LIDX_TABLE, query); + query = queryByLabelIndex(session, indexTable(), query); if (query == null) { return ImmutableList.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 ids = queryByLabelIndex(session, LIDX_TABLE, label); + Set 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 query( CassandraSessionPool.Session session, Query query) { - query = queryByLabelIndex(session, LIDX_TABLE, query); + query = queryByLabelIndex(session, indexTable(), query); if (query == null) { return ImmutableList.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); } diff --git a/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTablesWithMV.java b/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTablesWithMV.java index 2d870195d..30851ee47 100644 --- a/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTablesWithMV.java +++ b/hugegraph-scylladb/src/main/java/com/baidu/hugegraph/backend/store/scylladb/ScyllaDBTablesWithMV.java @@ -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); } } } diff --git a/hugegraph-test/pom.xml b/hugegraph-test/pom.xml index c2de11904..a9b9e3852 100644 --- a/hugegraph-test/pom.xml +++ b/hugegraph-test/pom.xml @@ -5,7 +5,7 @@ hugegraph com.baidu.hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT 4.0.0 diff --git a/hugegraph-test/src/main/java/com/baidu/hugegraph/unit/rocksdb/BaseRocksDBUnitTest.java b/hugegraph-test/src/main/java/com/baidu/hugegraph/unit/rocksdb/BaseRocksDBUnitTest.java index 11a58c86a..ba6ddbae3 100644 --- a/hugegraph-test/src/main/java/com/baidu/hugegraph/unit/rocksdb/BaseRocksDBUnitTest.java +++ b/hugegraph-test/src/main/java/com/baidu/hugegraph/unit/rocksdb/BaseRocksDBUnitTest.java @@ -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; } diff --git a/pom.xml b/pom.xml index 3d4e0fdf6..9713410be 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,7 @@ 4.0.0 com.baidu.hugegraph hugegraph - 0.7.0-SNAPSHOT + 0.7.1-SNAPSHOT pom 3.3.9