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

make create index label return async job if exists 

fixed #32

Change-Id: I8e23bd0dfcbe65d8f580f957143fcc1802c879f6
This commit is contained in:
zhoney 2018-08-27 18:57:11 +08:00 committed by zhoney
parent 44277abf80
commit 30fbc72a21
13 changed files with 280 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hugegraph-api/pom.xml
View File

@ -101,7 +101,7 @@
</addDefaultSpecificationEntries>
</manifest>
<manifestEntries>
<Implementation-Version>0.29.0.0</Implementation-Version>
<Implementation-Version>0.30.0.0</Implementation-Version>
</manifestEntries>
</archive>
</configuration>

29
hugegraph-api/src/main/java/com/baidu/hugegraph/api/job/TaskAPI.java
View File

@ -24,6 +24,7 @@ import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import javax.inject.Singleton;
import javax.ws.rs.BadRequestException;
@ -42,6 +43,7 @@ import org.slf4j.Logger;
import com.baidu.hugegraph.api.API;
import com.baidu.hugegraph.api.filter.StatusFilter.Status;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.backend.id.IdGenerator;
import com.baidu.hugegraph.core.GraphManager;
import com.baidu.hugegraph.server.RestServer;
@ -58,6 +60,7 @@ import com.google.common.collect.ImmutableMap;
public class TaskAPI extends API {
private static final Logger LOG = Log.logger(RestServer.class);
private static final long NO_LIMIT = -1L;
public static final String ACTION_CANCEL = "cancel";
@ -67,6 +70,7 @@ public class TaskAPI extends API {
public Map<String, List<Object>> list(@Context GraphManager manager,
@PathParam("graph") String graph,
@QueryParam("status") String status,
@QueryParam("ids") List<Long> ids,
@QueryParam("limit")
@DefaultValue("100") long limit) {
LOG.debug("Graph [{}] list tasks with status {}, limit {}",
@ -75,16 +79,35 @@ public class TaskAPI extends API {
TaskScheduler scheduler = graph(manager, graph).taskScheduler();
Iterator<HugeTask<Object>> itor;
if (status == null) {
itor = scheduler.findAllTask(limit);
if (!ids.isEmpty()) {
LOG.debug("Graph [{}] list tasks with ids {}, limit {}",
graph, ids, limit);
E.checkArgument(status == null,
"Not support status when query task by ids, " +
"but got status='%s'", status);
// Set limit to NO_LIMIT to ignore limit when query task by ids
limit = NO_LIMIT;
List<Id> idList = ids.stream().map(IdGenerator::of)
.collect(Collectors.toList());
itor = scheduler.tasks(idList);
} else {
itor = scheduler.findTask(parseStatus(status), limit);
LOG.debug("Graph [{}] list tasks with status {}, limit {}",
graph, status, limit);
if (status == null) {
itor = scheduler.findAllTask(limit);
} else {
itor = scheduler.findTask(parseStatus(status), limit);
}
}
List<Object> tasks = new ArrayList<>();
while (itor.hasNext()) {
tasks.add(itor.next().asMap(false));
}
if (limit != NO_LIMIT && tasks.size() > limit) {
tasks = tasks.subList(0, (int) limit);
}
return ImmutableMap.of("tasks", tasks);
}

4
hugegraph-api/src/main/java/com/baidu/hugegraph/api/schema/IndexLabelAPI.java
View File

@ -67,8 +67,8 @@ public class IndexLabelAPI extends API {
HugeGraph g = graph(manager, graph);
IndexLabel.Builder builder = jsonIndexLabel.convert2Builder(g);
IndexLabel indexLabel = builder.create();
return manager.serializer(g).writeIndexlabel(indexLabel);
IndexLabel.CreatedIndexLabel il = builder.createWithTask();
return manager.serializer(g).writeCreatedIndexLabel(il);
}
@GET

12
hugegraph-api/src/main/java/com/baidu/hugegraph/serializer/JsonSerializer.java
View File

@ -159,6 +159,18 @@ public class JsonSerializer implements Serializer {
return writeList("indexlabels", indexLabels);
}
@Override
public String writeCreatedIndexLabel(IndexLabel.CreatedIndexLabel cil) {
StringBuilder builder = new StringBuilder();
long id = cil.task() == null ? 0L : cil.task().asLong();
return builder.append("{\"index_label\": ")
.append(this.writeIndexlabel(cil.indexLabel()))
.append(", \"task_id\": ")
.append(id)
.append("}")
.toString();
}
@Override
public String writeVertex(Vertex vertex) {
return writeObject(vertex);

2
hugegraph-api/src/main/java/com/baidu/hugegraph/serializer/Serializer.java
View File

@ -52,6 +52,8 @@ public interface Serializer {
public String writeIndexlabels(List<IndexLabel> indexLabels);
public String writeCreatedIndexLabel(IndexLabel.CreatedIndexLabel cil);
public String writeVertex(Vertex v);
public String writeVertices(Iterator<Vertex> vertices, boolean paging);

3
hugegraph-api/src/main/java/com/baidu/hugegraph/version/ApiVersion.java
View File

@ -73,10 +73,11 @@ public final class ApiVersion {
* version 0.8:
* [0.28] Issue-153: Add task-cancel API
* [0.29] Issue-39: Add rays and rings RESTful API
* [0.30] Issue-32: Change index create API to return indexLabel and task id
*/
// The second parameter of Version.of() is for IDE running without JAR
public static final Version VERSION = Version.of(ApiVersion.class, "0.29");
public static final Version VERSION = Version.of(ApiVersion.class, "0.30");
public static final void check() {
// Check version of hugegraph-core. Firstly do check from version 0.3

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

@ -22,6 +22,7 @@ package com.baidu.hugegraph.backend.tx;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import com.baidu.hugegraph.HugeException;
import com.baidu.hugegraph.HugeGraph;
@ -52,6 +53,7 @@ import com.baidu.hugegraph.type.define.HugeKeys;
import com.baidu.hugegraph.type.define.SchemaStatus;
import com.baidu.hugegraph.util.E;
import com.baidu.hugegraph.util.LockUtil;
import com.google.common.collect.ImmutableSet;
public class SchemaTransaction extends IndexableTransaction {
@ -217,10 +219,15 @@ public class SchemaTransaction extends IndexableTransaction {
}
public Id rebuildIndex(SchemaElement schema) {
return this.rebuildIndex(schema, ImmutableSet.of());
}
public Id rebuildIndex(SchemaElement schema, Set<Id> dependencies) {
LOG.debug("SchemaTransaction rebuild index for {} with id '{}'",
schema.type(), schema.id());
SchemaCallable callable = new RebuildIndexCallable();
return asyncRun(this.graph(), schema.type(), schema.id(), callable);
return asyncRun(this.graph(), schema.type(), schema.id(), callable,
dependencies);
}
public void updateSchemaStatus(SchemaElement schema, SchemaStatus status) {
@ -352,10 +359,21 @@ public class SchemaTransaction extends IndexableTransaction {
private static Id asyncRun(HugeGraph graph, HugeType schemaType,
Id schemaId, SchemaCallable callable) {
String name = SchemaCallable.formatTaskName(schemaType, schemaId);
return asyncRun(graph, schemaType, schemaId,
callable, ImmutableSet.of());
}
private static Id asyncRun(HugeGraph graph, HugeType schemaType,
Id schemaId, SchemaCallable callable,
Set<Id> dependencies) {
String schemaName = graph.schemaTransaction()
.getSchema(schemaType, schemaId).name();
String name = SchemaCallable.formatTaskName(schemaType, schemaId,
schemaName);
JobBuilder<Object> builder = JobBuilder.of(graph).name(name)
.job(callable);
.job(callable)
.dependencies(dependencies);
HugeTask<?> task = builder.schedule();
// If SCHEMA_SYNC_DELETION is true, wait async thread done before

13
hugegraph-core/src/main/java/com/baidu/hugegraph/job/JobBuilder.java
View File

@ -19,6 +19,8 @@
package com.baidu.hugegraph.job;
import java.util.Set;
import com.baidu.hugegraph.HugeGraph;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.task.HugeTask;
@ -33,6 +35,7 @@ public class JobBuilder<T> {
private String name;
private String input;
private Job<T> job;
private Set<Id> dependencies;
public static <T> JobBuilder<T> of(final HugeGraph graph) {
return new JobBuilder<>(graph);
@ -57,6 +60,11 @@ public class JobBuilder<T> {
return this;
}
public JobBuilder<T> dependencies(Set<Id> dependencies) {
this.dependencies = dependencies;
return this;
}
public HugeTask<T> schedule() {
E.checkArgumentNotNull(this.name, "Job name can't be null");
E.checkArgumentNotNull(this.job, "Job can't be null");
@ -67,6 +75,11 @@ public class JobBuilder<T> {
if (this.input != null) {
task.input(this.input);
}
if (this.dependencies != null && !this.dependencies.isEmpty()) {
for (Id depend : this.dependencies) {
task.depends(depend);
}
}
TaskScheduler scheduler = this.graph.taskScheduler();
scheduler.schedule(task);

26
hugegraph-core/src/main/java/com/baidu/hugegraph/job/schema/SchemaCallable.java
View File

@ -17,32 +17,34 @@ public abstract class SchemaCallable extends Job<Object> {
public static final String REMOVE_SCHEMA = "remove_schema";
public static final String REBUILD_INDEX = "rebuild_index";
public static final String CREATE_INDEX = "create_index";
private static final String SPLITOR = ":";
protected HugeType schemaType() {
String name = this.task().name();
String[] parts = name.split(SPLITOR);
E.checkState(parts.length == 2 && parts[0] != null,
"Task name should be formatted to String 'TYPE:ID', " +
"but got '%s'", name);
String[] parts = name.split(SPLITOR, 3);
E.checkState(parts.length == 3 && parts[0] != null,
"Task name should be formatted to String " +
"'TYPE:ID:NAME', but got '%s'", name);
return HugeType.valueOf(parts[0]);
}
protected Id schemaId() {
String name = this.task().name();
String[] parts = name.split(SPLITOR);
E.checkState(parts.length == 2 && parts[1] != null,
"Task name should be formatted to String 'TYPE:ID', " +
"but got '%s'", name);
String[] parts = name.split(SPLITOR, 3);
E.checkState(parts.length == 3 && parts[1] != null,
"Task name should be formatted to String " +
"'TYPE:ID:NAME', but got '%s'", name);
return IdGenerator.of(Long.valueOf(parts[1]));
}
public static String formatTaskName(HugeType schemaType, Id schemaId) {
E.checkNotNull(schemaType, "schema type");
E.checkNotNull(schemaId, "schema id");
return String.join(SPLITOR, schemaType.toString(), schemaId.toString());
public static String formatTaskName(HugeType type, Id id, String name) {
E.checkNotNull(type, "schema type");
E.checkNotNull(id, "schema id");
E.checkNotNull(name, "schema name");
return String.join(SPLITOR, type.toString(), id.asString(), name);
}
protected static void removeIndexLabelFromBaseLabel(SchemaTransaction tx,

31
hugegraph-core/src/main/java/com/baidu/hugegraph/schema/IndexLabel.java
View File

@ -184,6 +184,8 @@ public class IndexLabel extends SchemaElement {
public interface Builder extends SchemaBuilder<IndexLabel> {
CreatedIndexLabel createWithTask();
Id rebuild();
Builder onV(String baseValue);
@ -202,4 +204,33 @@ public class IndexLabel extends SchemaElement {
Builder indexType(IndexType indexType);
}
public static class CreatedIndexLabel {
private IndexLabel indexLabel;
private Id task;
public CreatedIndexLabel(IndexLabel indexLabel, Id task) {
E.checkNotNull(indexLabel, "index label");
this.indexLabel = indexLabel;
this.task = task;
}
public void indexLabel(IndexLabel indexLabel) {
E.checkNotNull(indexLabel, "index label");
this.indexLabel = indexLabel;
}
public IndexLabel indexLabel() {
return this.indexLabel;
}
public void task(Id task) {
this.task = task;
}
public Id task() {
return this.task;
}
}
}

55
hugegraph-core/src/main/java/com/baidu/hugegraph/schema/builder/IndexLabelBuilder.java
View File

@ -21,14 +21,17 @@ package com.baidu.hugegraph.schema.builder;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeoutException;
import com.baidu.hugegraph.HugeException;
import com.baidu.hugegraph.HugeGraph;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.backend.tx.GraphTransaction;
import com.baidu.hugegraph.backend.tx.SchemaTransaction;
import com.baidu.hugegraph.config.CoreOptions;
import com.baidu.hugegraph.config.HugeConfig;
import com.baidu.hugegraph.exception.ExistedException;
import com.baidu.hugegraph.exception.NotSupportException;
import com.baidu.hugegraph.schema.EdgeLabel;
@ -37,6 +40,7 @@ import com.baidu.hugegraph.schema.PropertyKey;
import com.baidu.hugegraph.schema.SchemaElement;
import com.baidu.hugegraph.schema.SchemaLabel;
import com.baidu.hugegraph.schema.VertexLabel;
import com.baidu.hugegraph.task.TaskScheduler;
import com.baidu.hugegraph.type.HugeType;
import com.baidu.hugegraph.type.define.Cardinality;
import com.baidu.hugegraph.type.define.DataType;
@ -44,6 +48,7 @@ import com.baidu.hugegraph.type.define.IndexType;
import com.baidu.hugegraph.type.define.SchemaStatus;
import com.baidu.hugegraph.util.CollectionUtil;
import com.baidu.hugegraph.util.E;
import com.baidu.hugegraph.util.InsertionOrderUtil;
public class IndexLabelBuilder implements IndexLabel.Builder {
@ -83,7 +88,7 @@ public class IndexLabelBuilder implements IndexLabel.Builder {
}
@Override
public IndexLabel create() {
public IndexLabel.CreatedIndexLabel createWithTask() {
SchemaElement.checkName(this.name,
this.transaction.graph().configuration());
IndexLabel indexLabel = this.transaction.getIndexLabel(this.name);
@ -91,7 +96,7 @@ public class IndexLabelBuilder implements IndexLabel.Builder {
if (this.checkExist) {
throw new ExistedException("index label", this.name);
}
return indexLabel;
return new IndexLabel.CreatedIndexLabel(indexLabel, null);
}
SchemaLabel schemaLabel = this.loadElement();
@ -105,7 +110,7 @@ public class IndexLabelBuilder implements IndexLabel.Builder {
// Delete index label which is prefix of the new index label
// TODO: use event to replace direct call
this.removeSubIndex(schemaLabel);
Set<Id> removeTasks = this.removeSubIndex(schemaLabel);
// Create index label
indexLabel = this.build();
@ -113,8 +118,29 @@ public class IndexLabelBuilder implements IndexLabel.Builder {
this.transaction.addIndexLabel(schemaLabel, indexLabel);
// TODO: use event to replace direct call
this.rebuildIndexIfNeeded(schemaLabel, indexLabel);
Id rebuildTask = this.rebuildIndexIfNeeded(schemaLabel, indexLabel,
removeTasks);
return new IndexLabel.CreatedIndexLabel(indexLabel, rebuildTask);
}
@Override
public IndexLabel create() {
IndexLabel.CreatedIndexLabel createdIndexLabel = this.createWithTask();
Id task = createdIndexLabel.task();
IndexLabel indexLabel = createdIndexLabel.indexLabel();
if (task == null) {
E.checkNotNull(indexLabel, "index label");
return indexLabel;
}
TaskScheduler scheduler = this.transaction.graph().taskScheduler();
try {
// TODO: read timeout from config file
scheduler.waitUntilTaskCompleted(task, 30L);
} catch (TimeoutException e) {
throw new HugeException(
"Failed to wait index-creating task completed", e);
}
return indexLabel;
}
@ -316,8 +342,8 @@ public class IndexLabelBuilder implements IndexLabel.Builder {
}
}
private void removeSubIndex(SchemaLabel schemaLabel) {
HashSet<Id> overrideIndexLabelIds = new HashSet<>();
private Set<Id> removeSubIndex(SchemaLabel schemaLabel) {
Set<Id> overrideIndexLabelIds = InsertionOrderUtil.newSet();
for (Id id : schemaLabel.indexLabels()) {
IndexLabel old = this.transaction.getIndexLabel(id);
if (this.indexType != old.indexType()) {
@ -334,14 +360,19 @@ public class IndexLabelBuilder implements IndexLabel.Builder {
overrideIndexLabelIds.add(id);
}
}
Set<Id> tasks = InsertionOrderUtil.newSet();
for (Id id : overrideIndexLabelIds) {
schemaLabel.removeIndexLabel(id);
this.transaction.removeIndexLabel(id);
Id task = this.transaction.removeIndexLabel(id);
E.checkNotNull(task, "remove sub index label task");
tasks.add(task);
}
return tasks;
}
private void rebuildIndexIfNeeded(SchemaLabel schemaLabel,
IndexLabel indexLabel) {
private Id rebuildIndexIfNeeded(SchemaLabel schemaLabel,
IndexLabel indexLabel,
Set<Id> dependencies) {
GraphTransaction tx = this.transaction.graph().graphTransaction();
boolean needRebuild;
if (this.baseType == HugeType.VERTEX_LABEL) {
@ -352,12 +383,14 @@ public class IndexLabelBuilder implements IndexLabel.Builder {
needRebuild = tx.queryEdgesByLabel((EdgeLabel) schemaLabel, 1L)
.hasNext();
}
Id task = null;
if (needRebuild) {
// rebuildIndex() will set status to CREATED after REBUILDING
this.transaction.rebuildIndex(indexLabel);
task = this.transaction.rebuildIndex(indexLabel, dependencies);
} else {
this.transaction.updateSchemaStatus(indexLabel,
SchemaStatus.CREATED);
}
return task;
}
}

94
hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTask.java
View File

@ -26,7 +26,10 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.FutureTask;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
import org.apache.tinkerpop.gremlin.structure.T;
@ -35,8 +38,10 @@ import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.slf4j.Logger;
import com.baidu.hugegraph.backend.id.Id;
import com.baidu.hugegraph.backend.id.IdGenerator;
import com.baidu.hugegraph.type.define.SerialEnum;
import com.baidu.hugegraph.util.E;
import com.baidu.hugegraph.util.InsertionOrderUtil;
import com.baidu.hugegraph.util.Log;
public class HugeTask<V> extends FutureTask<V> {
@ -49,7 +54,7 @@ public class HugeTask<V> extends FutureTask<V> {
private String name;
private final Id id;
private final Id parent;
private List<Id> children;
private Set<Id> dependencies;
private String description;
private Date create;
private volatile TaskStatus status;
@ -76,7 +81,7 @@ public class HugeTask<V> extends FutureTask<V> {
this.name = null;
this.id = id;
this.parent = parent;
this.children = null;
this.dependencies = null;
this.description = null;
this.status = TaskStatus.NEW;
this.progress = 0;
@ -95,15 +100,17 @@ public class HugeTask<V> extends FutureTask<V> {
return this.parent;
}
public List<Id> children() {
return Collections.unmodifiableList(this.children);
public Set<Id> dependencies() {
return Collections.unmodifiableSet(this.dependencies);
}
public void child(Id id) {
if (this.children == null) {
this.children = new ArrayList<>();
public void depends(Id id) {
E.checkState(this.status == TaskStatus.NEW,
"Can't add dependency in status '%s'", this.status);
if (this.dependencies == null) {
this.dependencies = InsertionOrderUtil.newSet();
}
this.children.add(id);
this.dependencies.add(id);
}
public TaskStatus status() {
@ -190,8 +197,10 @@ public class HugeTask<V> extends FutureTask<V> {
@Override
public void run() {
assert this.status.code() < TaskStatus.RUNNING.code();
this.status(TaskStatus.RUNNING);
super.run();
if (this.checkDependenciesSuccess()) {
this.status(TaskStatus.RUNNING);
super.run();
}
}
@Override
@ -241,6 +250,35 @@ public class HugeTask<V> extends FutureTask<V> {
super.setException(e);
}
protected boolean checkDependenciesSuccess() {
if (this.dependencies == null || this.dependencies.isEmpty()) {
return true;
}
for (Id dependency : this.dependencies) {
HugeTask task = this.callable.scheduler().task(dependency);
if (!task.completed()) {
// Dependent task not completed, re-schedule self
this.callable.scheduler().schedule(this);
return false;
} else if (task.status() == TaskStatus.CANCELLED) {
this.status(TaskStatus.CANCELLED);
this.result = String.format(
"Cancelled due to dependent task '%s' cancelled",
dependency);
this.done();
return false;
} else if (task.status() == TaskStatus.FAILED) {
this.status(TaskStatus.FAILED);
this.result = String.format(
"Failed due to dependent task '%s' failed",
dependency);
this.done();
return false;
}
}
return true;
}
protected TaskCallable<V> callable() {
return this.callable;
}
@ -258,8 +296,8 @@ public class HugeTask<V> extends FutureTask<V> {
case P.NAME:
this.name = (String) value;
break;
case P.DESCRIPTION:
this.description = (String) value;
case P.CALLABLE:
// pass
break;
case P.STATUS:
this.status(SerialEnum.fromCode(TaskStatus.class, (byte) value));
@ -270,11 +308,20 @@ public class HugeTask<V> extends FutureTask<V> {
case P.CREATE:
this.create = (Date) value;
break;
case P.RETRIES:
this.retries = (int) value;
break;
case P.DESCRIPTION:
this.description = (String) value;
break;
case P.UPDATE:
this.update = (Date) value;
break;
case P.RETRIES:
this.retries = (int) value;
case P.DEPENDENCIES:
@SuppressWarnings("unchecked")
Set<Long> values = (Set<Long>) value;
this.dependencies = values.stream().map(IdGenerator::of)
.collect(toOrderSet());
break;
case P.INPUT:
this.input = (String) value;
@ -282,9 +329,6 @@ public class HugeTask<V> extends FutureTask<V> {
case P.RESULT:
this.result = (String) value;
break;
case P.CALLABLE:
// pass
break;
default:
throw new AssertionError("Unsupported key: " + key);
}
@ -333,6 +377,12 @@ public class HugeTask<V> extends FutureTask<V> {
list.add(this.update);
}
if (this.dependencies != null) {
list.add(P.DEPENDENCIES);
list.add(this.dependencies.stream().map(Id::asLong)
.collect(toOrderSet()));
}
if (this.input != null) {
list.add(P.INPUT);
list.add(this.input);
@ -372,6 +422,11 @@ public class HugeTask<V> extends FutureTask<V> {
if (this.update != null) {
map.put(Hidden.unHide(P.UPDATE), this.update);
}
if (this.dependencies != null) {
Set<Long> value = this.dependencies.stream().map(Id::asLong)
.collect(toOrderSet());
map.put(Hidden.unHide(P.DEPENDENCIES), value);
}
if (withDetails && this.input != null) {
map.put(Hidden.unHide(P.INPUT), this.input);
}
@ -400,6 +455,10 @@ public class HugeTask<V> extends FutureTask<V> {
return task;
}
private static <T> Collector<T, ?, Set<T>> toOrderSet() {
return Collectors.toCollection(InsertionOrderUtil::newSet);
}
public static final class P {
public static final String TASK = Hidden.hide("task");
@ -418,6 +477,7 @@ public class HugeTask<V> extends FutureTask<V> {
public static final String RETRIES = "~task_retries";
public static final String INPUT = "~task_input";
public static final String RESULT = "~task_result";
public static final String DEPENDENCIES = "~task_dependencies";
//public static final String PARENT = hide("parent");
//public static final String CHILDREN = hide("children");

36
hugegraph-core/src/main/java/com/baidu/hugegraph/task/TaskScheduler.java
View File

@ -41,6 +41,7 @@ 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.exception.NotFoundException;
import com.baidu.hugegraph.iterator.ExtendableIterator;
import com.baidu.hugegraph.iterator.MapperIterator;
import com.baidu.hugegraph.schema.IndexLabel;
import com.baidu.hugegraph.schema.PropertyKey;
@ -203,6 +204,29 @@ public class TaskScheduler {
return this.findTask(id);
}
public <V> Iterator<HugeTask<V>> tasks(List<Id> ids) {
List<Id> taskIdsNotInMem = new ArrayList<>();
List<HugeTask<V>> taskInMem = new ArrayList<>();
for (Id id : ids) {
@SuppressWarnings("unchecked")
HugeTask<V> task = (HugeTask<V>) this.tasks.get(id);
if (task != null) {
taskInMem.add(task);
} else {
taskIdsNotInMem.add(id);
}
}
@SuppressWarnings("unchecked")
ExtendableIterator<HugeTask<V>> iterator;
if (taskInMem.isEmpty()) {
iterator = new ExtendableIterator<>();
} else {
iterator = new ExtendableIterator<>(taskInMem.iterator());
}
iterator.extend(this.findTasks(taskIdsNotInMem));
return iterator;
}
public <V> HugeTask<V> findTask(Id id) {
HugeTask<V> result = this.submit(() -> {
HugeTask<V> task = null;
@ -219,6 +243,14 @@ public class TaskScheduler {
return result;
}
public <V> Iterator<HugeTask<V>> findTasks(List<Id> ids) {
Object[] idArray = ids.toArray(new Id[ids.size()]);
return this.submit(() -> {
Iterator<Vertex> vertices = this.tx().queryVertices(idArray);
return new MapperIterator<>(vertices, HugeTask::fromVertex);
});
}
public <V> Iterator<HugeTask<V>> findAllTask(long limit) {
return this.queryTask(ImmutableMap.of(), limit);
}
@ -352,7 +384,7 @@ public class TaskScheduler {
.properties(properties)
.useCustomizeNumberId()
.nullableKeys(P.DESCRIPTION, P.UPDATE,
P.INPUT, P.RESULT)
P.INPUT, P.RESULT, P.DEPENDENCIES)
.enableLabelIndex(true)
.build();
graph.schemaTransaction().addVertexLabel(label);
@ -375,6 +407,8 @@ public class TaskScheduler {
props.add(createPropertyKey(P.RETRIES, DataType.INT));
props.add(createPropertyKey(P.INPUT));
props.add(createPropertyKey(P.RESULT));
props.add(createPropertyKey(P.DEPENDENCIES, DataType.LONG,
Cardinality.SET));
return props.toArray(new String[0]);
}