You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by dk...@apache.org on 2015/09/01 16:26:18 UTC
[02/19] incubator-tinkerpop git commit: implemented
BulkLoaderVertexProgram and 2 default BulkLoader implementations
implemented BulkLoaderVertexProgram and 2 default BulkLoader implementations
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/048fcdc9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/048fcdc9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/048fcdc9
Branch: refs/heads/tp30
Commit: 048fcdc9caa4467aa7e065f525f386db3991ad75
Parents: 7a9b075
Author: Daniel Kuppitz <da...@hotmail.com>
Authored: Mon Aug 24 02:28:55 2015 +0200
Committer: Daniel Kuppitz <da...@hotmail.com>
Committed: Mon Aug 24 02:28:55 2015 +0200
----------------------------------------------------------------------
.../computer/bulkloading/BulkLoader.java | 105 ++++++++
.../bulkloading/BulkLoaderVertexProgram.java | 267 +++++++++++++++++++
.../computer/bulkloading/DefaultBulkLoader.java | 89 +++++++
.../bulkloading/IncrementalBulkLoader.java | 78 ++++++
.../gremlin/structure/io/gryo/GryoMapper.java | 19 +-
.../structure/io/gryo/PairSerializer.java | 42 +++
6 files changed, 591 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/048fcdc9/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoader.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoader.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoader.java
new file mode 100644
index 0000000..c6a3cc1
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoader.java
@@ -0,0 +1,105 @@
+/*
+ * 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 org.apache.tinkerpop.gremlin.process.computer.bulkloading;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.VertexProperty;
+
+/**
+ * @author Daniel Kuppitz (http://gremlin.guru)
+ */
+public interface BulkLoader {
+
+ /**
+ * Gets or creates a clone of the given vertex in the given graph.
+ *
+ * @param vertex The vertex to be cloned.
+ * @param graph The graph that holds the cloned vertex after this method was called.
+ * @param g A standard traversal source for the given graph.
+ * @return The cloned vertex.
+ */
+ public Vertex getOrCreateVertex(final Vertex vertex, final Graph graph, final GraphTraversalSource g);
+
+ /**
+ * Gets or creates a clone of the given edge between the given in- and out-vertices.
+ *
+ * @param edge The edge to be cloned.
+ * @param outVertex The out-vertex in the given graph..
+ * @param inVertex The in-vertex in the given graph.
+ * @param graph The graph that holds the cloned edge after this method was called.
+ * @param g A standard traversal source for the given graph.
+ * @return The cloned edge.
+ */
+ public Edge getOrCreateEdge(final Edge edge, final Vertex outVertex, final Vertex inVertex, final Graph graph, final GraphTraversalSource g);
+
+ /**
+ * Gets or creates a clone of the given property for the given vertex.
+ *
+ * @param property The property to be cloned.
+ * @param vertex The vertex in the given graph..
+ * @param graph The graph that holds the given vertex.
+ * @param g A standard traversal source for the given graph.
+ * @return The cloned property.
+ */
+ public VertexProperty getOrCreateVertexProperty(final VertexProperty<?> property, final Vertex vertex, final Graph graph, final GraphTraversalSource g);
+
+ /**
+ * Get a vertex that matches the given vertex from the given graph.
+ *
+ * @param vertex The vertex to be matched.
+ * @param graph The graph that holds the given vertex.
+ * @param g A standard traversal source for the given graph.
+ * @return The matched vertex.
+ */
+ public default Vertex getVertex(final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
+ return useUserSuppliedIds()
+ ? getVertexById(vertex.id(), graph, g)
+ : g.V().has(BulkLoaderVertexProgram.BULK_LOADER_VERTEX_ID, vertex.id()).next();
+ }
+
+ /**
+ * Gets a vertex by its ID from the given graph.
+ *
+ * @param id The vertex ID.
+ * @param graph The graph that holds the vertex with the given ID.
+ * @param g A standard traversal source for the given graph.
+ * @return The vertex with the given ID.
+ */
+ public default Vertex getVertexById(final Object id, final Graph graph, final GraphTraversalSource g) {
+ return graph.vertices(id).next();
+ }
+
+ /**
+ * @return Whether to use user supplied identifiers or not.
+ */
+ public boolean useUserSuppliedIds();
+
+ /**
+ * @return Whether original vertex identifiers are stored in the target graph or not.
+ */
+ public default boolean storeOriginalIds() {
+ return !useUserSuppliedIds();
+ }
+
+ public void configure(final Configuration configuration);
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/048fcdc9/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgram.java
new file mode 100644
index 0000000..4fd017e
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgram.java
@@ -0,0 +1,267 @@
+/*
+ * 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 org.apache.tinkerpop.gremlin.process.computer.bulkloading;
+
+import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
+import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
+import org.apache.tinkerpop.gremlin.process.computer.Messenger;
+import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
+import org.apache.tinkerpop.gremlin.structure.Direction;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.VertexProperty;
+import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
+import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
+import org.javatuples.Pair;
+import org.javatuples.Tuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.FileInputStream;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * @author Daniel Kuppitz (http://gremlin.guru)
+ */
+public class BulkLoaderVertexProgram implements VertexProgram<Tuple> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BulkLoaderVertexProgram.class);
+
+ private static final String BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX = "bulkloader.conf";
+ private static final String BULK_LOADER_CLASS = BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX + ".class";
+ private static final String BULK_LOADER_CFG_PREFIX = BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX + ".loader";
+
+ public static final String BULK_LOADER_VERTEX_ID = BulkLoaderVertexProgram.class.getCanonicalName() + "bulkloader.vertex-id";
+
+ private MessageScope messageScope = MessageScope.Local.of(__::inE);
+ private Configuration configuration;
+ private Graph graph;
+ private GraphTraversalSource g;
+ private BulkLoader bulkLoader;
+ private Set<String> elementComputeKeys = new HashSet<String>() {{
+ add(BULK_LOADER_VERTEX_ID);
+ }};
+
+ private BulkLoaderVertexProgram() {
+
+ }
+
+ @Override
+ public void setup(final Memory memory) {
+
+ }
+
+ @Override
+ public void loadState(final Graph graph, final Configuration config) {
+ configuration = new BaseConfiguration();
+ if (config != null) {
+ final Iterator<String> keys = config.getKeys();
+ while (keys.hasNext()) {
+ final String key = keys.next();
+ configuration.setProperty(key, config.getProperty(key));
+ }
+ }
+ final Configuration blvpConfiguration = graph.configuration().subset(BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX);
+ blvpConfiguration.getKeys().forEachRemaining(key -> configuration.setProperty(key, blvpConfiguration.getProperty(key)));
+ }
+
+ @Override
+ public void storeState(final Configuration config) {
+ VertexProgram.super.storeState(config);
+ if (configuration != null) {
+ configuration.getKeys().forEachRemaining(key -> config.setProperty(key, configuration.getProperty(key)));
+ }
+ }
+
+ @Override
+ public void workerIterationStart(final Memory memory) {
+ if (null == graph) {
+ graph = GraphFactory.open(configuration);
+ LOGGER.info("Opened Graph instance: {}", graph);
+ if (!graph.features().graph().supportsConcurrentAccess()) {
+ throw new IllegalStateException("The given graph instance does not allow concurrent access.");
+ }
+ if (graph.features().graph().supportsTransactions()) {
+ if (!graph.features().graph().supportsThreadedTransactions()) {
+ throw new IllegalStateException("The given graph instance does not support threaded transactions.");
+ }
+ }
+ g = graph.traversal();
+ final String bulkLoaderClassName = configuration.getString(BULK_LOADER_CLASS, DefaultBulkLoader.class.getCanonicalName());
+ try {
+ final Class<?> bulkLoaderClass = Class.forName(bulkLoaderClassName);
+ bulkLoader = (BulkLoader) bulkLoaderClass.getConstructor().newInstance();
+ } catch (ClassNotFoundException e) {
+ LOGGER.error("Unable to find custom bulk loader class: {}", bulkLoaderClassName);
+ throw new IllegalStateException(e);
+ } catch (Exception e) {
+ LOGGER.error("Unable to create an instance of the given bulk loader class: {}", bulkLoaderClassName);
+ throw new IllegalStateException(e);
+ }
+ bulkLoader.configure(configuration.subset(BULK_LOADER_CFG_PREFIX));
+ } else {
+ LOGGER.warn("Leaked Graph instance: {}", graph);
+ }
+ }
+
+ @Override
+ public void workerIterationEnd(final Memory memory) {
+ if (null != graph) {
+ if (graph.features().graph().supportsTransactions()) {
+ LOGGER.info("Committing transaction on Graph instance: {}", graph);
+ graph.tx().commit(); // TODO will Giraph/MR restart the program and re-run execute if this fails?
+ LOGGER.debug("Committed transaction on Graph instance: {}", graph);
+ }
+ try {
+ graph.close();
+ LOGGER.info("Closed Graph instance: {}", graph);
+ graph = null;
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close Graph instance", e);
+ }
+ }
+ }
+
+ @Override
+ public void execute(final Vertex sourceVertex, final Messenger<Tuple> messenger, final Memory memory) {
+ if (memory.isInitialIteration()) {
+ // get or create the vertex
+ final Vertex targetVertex = bulkLoader.getOrCreateVertex(sourceVertex, graph, g);
+ // write all the properties of the vertex to the newly created vertex
+ final Iterator<VertexProperty<Object>> vpi = sourceVertex.properties();
+ while (vpi.hasNext()) {
+ bulkLoader.getOrCreateVertexProperty(vpi.next(), targetVertex, graph, g);
+ }
+ if (!bulkLoader.useUserSuppliedIds()) {
+ // create an id pair and send it to all the vertex's incoming adjacent vertices
+ sourceVertex.property(BULK_LOADER_VERTEX_ID, targetVertex.id());
+ messenger.sendMessage(this.messageScope, Pair.with(sourceVertex.id(), targetVertex.id()));
+ }
+ } else {
+ if (bulkLoader.useUserSuppliedIds()) {
+ final Vertex outV = bulkLoader.getVertex(sourceVertex, graph, g);
+ sourceVertex.edges(Direction.OUT).forEachRemaining(edge -> {
+ final Vertex inV = bulkLoader.getVertex(edge.inVertex(), graph, g);
+ bulkLoader.getOrCreateEdge(edge, outV, inV, graph, g);
+ });
+ } else {
+ // create an id map and populate it with all the incoming messages
+ final Map<Object, Object> idPairs = new HashMap<>();
+ final Iterator<Tuple> idi = messenger.receiveMessages();
+ while (idi.hasNext()) {
+ final Tuple idPair = idi.next();
+ idPairs.put(idPair.getValue(0), idPair.getValue(1));
+ }
+ // get the vertex given the dummy id property
+ final Long outVId = sourceVertex.value(BULK_LOADER_VERTEX_ID);
+ final Vertex outV = bulkLoader.getVertexById(outVId, graph, g);
+ // for all the incoming edges of the vertex, get the incoming adjacent vertex and write the edge and its properties
+ sourceVertex.edges(Direction.OUT).forEachRemaining(edge -> {
+ final Object inVId = idPairs.get(edge.inVertex().id());
+ final Vertex inV = bulkLoader.getVertexById(inVId, graph, g);
+ bulkLoader.getOrCreateEdge(edge, outV, inV, graph, g);
+ });
+ }
+ }
+ }
+
+ @Override
+ public boolean terminate(final Memory memory) {
+ return memory.getIteration() >= 1;
+ }
+
+ @Override
+ public Set<String> getElementComputeKeys() {
+ return elementComputeKeys;
+ }
+
+ @Override
+ public Set<MessageScope> getMessageScopes(final Memory memory) {
+ return new HashSet<MessageScope>() {{
+ add(messageScope);
+ }};
+ }
+
+ @Override
+ public VertexProgram<Tuple> clone() {
+ return this;
+ }
+
+ @Override
+ public GraphComputer.ResultGraph getPreferredResultGraph() {
+ return GraphComputer.ResultGraph.NEW;
+ }
+
+ @Override
+ public GraphComputer.Persist getPreferredPersist() {
+ return GraphComputer.Persist.EDGES;
+ }
+
+ @Override
+ public String toString() {
+ return StringFactory.vertexProgramString(this, "");
+ }
+
+ public static Builder build() {
+ return new Builder();
+ }
+
+ public static class Builder extends AbstractVertexProgramBuilder<Builder> {
+
+ private Builder() {
+ super(BulkLoaderVertexProgram.class);
+ }
+
+ public Builder graphConfiguration(final String propertiesFileLocation) {
+ try {
+ final Properties properties = new Properties();
+ properties.load(new FileInputStream(propertiesFileLocation));
+ properties.forEach((key, value) -> configuration.setProperty(BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX + "." + key, value));
+ return this;
+ } catch (final Exception e) {
+ throw new IllegalArgumentException(e.getMessage(), e);
+ }
+ }
+
+ public Builder bulkLoaderClass(final Class<? extends BulkLoader> bulkLoaderClass) {
+ return bulkLoaderClass(bulkLoaderClass, null);
+ }
+
+ public Builder bulkLoaderClass(final Class<? extends BulkLoader> bulkLoaderClass, final Configuration configuration) {
+ this.configuration.setProperty(BULK_LOADER_CLASS, bulkLoaderClass.getCanonicalName());
+ if (configuration != null) {
+ configuration.getKeys().forEachRemaining(key -> this.configuration.addProperty(
+ BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX + "." + key, configuration.getProperty(key)
+ ));
+ }
+ return this;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/048fcdc9/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/DefaultBulkLoader.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/DefaultBulkLoader.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/DefaultBulkLoader.java
new file mode 100644
index 0000000..a1aaf92
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/DefaultBulkLoader.java
@@ -0,0 +1,89 @@
+/*
+ * 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 org.apache.tinkerpop.gremlin.process.computer.bulkloading;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.T;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.VertexProperty;
+
+/**
+ * @author Daniel Kuppitz (http://gremlin.guru)
+ */
+public class DefaultBulkLoader implements BulkLoader {
+
+ private boolean storeOriginalIds = false;
+ private boolean useUserSuppliedIds = false;
+
+ @Override
+ public Vertex getOrCreateVertex(final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
+ if (useUserSuppliedIds()) {
+ return graph.addVertex(T.id, vertex.id(), T.label, vertex.label());
+ }
+ final Vertex v = graph.addVertex(T.label, vertex.label());
+ if (storeOriginalIds()) {
+ v.property(BulkLoaderVertexProgram.BULK_LOADER_VERTEX_ID, vertex.id());
+ }
+ return v;
+ }
+
+ @Override
+ public Edge getOrCreateEdge(final Edge edge, final Vertex outVertex, final Vertex inVertex, final Graph graph, final GraphTraversalSource g) {
+ final Edge e = outVertex.addEdge(edge.label(), inVertex);
+ edge.properties().forEachRemaining(property -> e.property(property.key(), property.value()));
+ return e;
+ }
+
+ @Override
+ public VertexProperty getOrCreateVertexProperty(final VertexProperty<?> property, final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
+ final VertexProperty<?> vp = vertex.property(property.key(), property.value());
+ vp.properties().forEachRemaining(metaProperty -> property.property(metaProperty.key(), metaProperty.value()));
+ return vp;
+ }
+
+ @Override
+ public Vertex getVertex(final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
+ return useUserSuppliedIds()
+ ? getVertexById(vertex.id(), graph, g)
+ : g.V().has(vertex.label(), BulkLoaderVertexProgram.BULK_LOADER_VERTEX_ID, vertex.id()).next();
+ }
+
+ @Override
+ public boolean useUserSuppliedIds() {
+ return useUserSuppliedIds;
+ }
+
+ @Override
+ public boolean storeOriginalIds() {
+ return storeOriginalIds;
+ }
+
+ @Override
+ public void configure(final Configuration configuration) {
+ if (configuration.containsKey("use-user-supplied-ids")) {
+ useUserSuppliedIds = configuration.getBoolean("use-user-supplied-ids");
+ }
+ if (configuration.containsKey("store-original-ids")) {
+ storeOriginalIds = configuration.getBoolean("store-original-ids");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/048fcdc9/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/IncrementalBulkLoader.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/IncrementalBulkLoader.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/IncrementalBulkLoader.java
new file mode 100644
index 0000000..23af2e2
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/IncrementalBulkLoader.java
@@ -0,0 +1,78 @@
+/*
+ * 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 org.apache.tinkerpop.gremlin.process.computer.bulkloading;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.NoSuchElementException;
+
+/**
+ * @author Daniel Kuppitz (http://gremlin.guru)
+ */
+public class IncrementalBulkLoader extends DefaultBulkLoader {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(IncrementalBulkLoader.class);
+
+ @Override
+ public Vertex getOrCreateVertex(final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
+ if (useUserSuppliedIds()) {
+ try {
+ return getVertexById(vertex.id(), graph, g);
+ } catch (NoSuchElementException ignored) {
+ }
+ } else {
+ final Traversal<Vertex, Vertex> t = g.V().has(vertex.label(), BulkLoaderVertexProgram.BULK_LOADER_VERTEX_ID, vertex.id());
+ if (t.hasNext()) return t.next();
+ }
+ return super.getOrCreateVertex(vertex, graph, g);
+ }
+
+ @Override
+ public Edge getOrCreateEdge(final Edge edge, final Vertex outVertex, final Vertex inVertex, final Graph graph, final GraphTraversalSource g) {
+ final Traversal<Vertex, Edge> t = g.V(outVertex).outE(edge.label()).filter(__.inV().is(inVertex));
+ if (t.hasNext()) {
+ final Edge e = t.next();
+ edge.properties().forEachRemaining(property -> e.property(property.key(), property.value()));
+ return e;
+ }
+ return super.getOrCreateEdge(edge, outVertex, inVertex, graph, g);
+ }
+
+ @Override
+ public boolean storeOriginalIds() {
+ return !useUserSuppliedIds();
+ }
+
+ @Override
+ public void configure(final Configuration configuration) {
+ super.configure(configuration);
+ if (configuration.containsKey("store-original-ids")) {
+ LOGGER.warn("{} automatically determines whether original identifiers should be stored or not, hence the " +
+ "configuration setting 'store-original-ids' will be ignored.", this.getClass());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/048fcdc9/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java
index 8b96052..0dc5d44 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java
@@ -20,6 +20,7 @@ package org.apache.tinkerpop.gremlin.structure.io.gryo;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
+import org.apache.tinkerpop.gremlin.process.traversal.Contains;
import org.apache.tinkerpop.gremlin.process.traversal.Path;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.Tree;
@@ -31,7 +32,6 @@ import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSe
import org.apache.tinkerpop.gremlin.process.traversal.util.DependantMutableMetrics;
import org.apache.tinkerpop.gremlin.process.traversal.util.MutableMetrics;
import org.apache.tinkerpop.gremlin.process.traversal.util.StandardTraversalMetrics;
-import org.apache.tinkerpop.gremlin.process.traversal.Contains;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Graph;
@@ -96,12 +96,12 @@ import java.util.stream.Collectors;
* serializers in one of three ways:
* <p/>
* <ol>
- * <li>Register just the custom class with a {@code null} {@link Serializer} implementation</li>
- * <li>Register the custom class with a {@link Serializer} implementation</li>
- * <li>
- * Register the custom class with a {@code Function<Kryo, Serializer>} for those cases where the
- * {@link Serializer} requires the {@link Kryo} instance to get constructed.
- * </li>
+ * <li>Register just the custom class with a {@code null} {@link Serializer} implementation</li>
+ * <li>Register the custom class with a {@link Serializer} implementation</li>
+ * <li>
+ * Register the custom class with a {@code Function<Kryo, Serializer>} for those cases where the
+ * {@link Serializer} requires the {@link Kryo} instance to get constructed.
+ * </li>
* </ol>
* <p/>
* For example:
@@ -223,7 +223,7 @@ public final class GryoMapper implements Mapper<Kryo> {
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ReferenceVertex.class, null, 84));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ReferencePath.class, null, 85));
- add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(StarGraph.class, kryo -> StarGraphGryoSerializer.with(Direction.BOTH), 86)); // ***LAST ID**
+ add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(StarGraph.class, kryo -> StarGraphGryoSerializer.with(Direction.BOTH), 86));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Edge.class, kryo -> new GryoSerializers.EdgeSerializer(), 65));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Vertex.class, kryo -> new GryoSerializers.VertexSerializer(), 66));
@@ -246,6 +246,7 @@ public final class GryoMapper implements Mapper<Kryo> {
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(MapReduce.NullObject.class, null, 74));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(AtomicLong.class, null, 79));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(DependantMutableMetrics.class, null, 80));
+ add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Pair.class, kryo -> new PairSerializer(), 87)); // ***LAST ID**
}};
private List<IoRegistry> registries = new ArrayList<>();
@@ -300,7 +301,7 @@ public final class GryoMapper implements Mapper<Kryo> {
*/
public GryoMapper create() {
// consult the registry if provided and inject registry entries as custom classes.
- registries.forEach(registry-> {
+ registries.forEach(registry -> {
final List<Pair<Class, Object>> serializers = registry.find(GryoIo.class);
serializers.forEach(p -> {
if (null == p.getValue1())
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/048fcdc9/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/PairSerializer.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/PairSerializer.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/PairSerializer.java
new file mode 100644
index 0000000..e5e92e7
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/PairSerializer.java
@@ -0,0 +1,42 @@
+/*
+ * 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 org.apache.tinkerpop.gremlin.structure.io.gryo;
+
+import org.apache.tinkerpop.shaded.kryo.Kryo;
+import org.apache.tinkerpop.shaded.kryo.Serializer;
+import org.apache.tinkerpop.shaded.kryo.io.Input;
+import org.apache.tinkerpop.shaded.kryo.io.Output;
+import org.javatuples.Pair;
+
+/**
+ * @author Daniel Kuppitz (http://gremlin.guru)
+ */
+final class PairSerializer extends Serializer<Pair> {
+
+ @Override
+ public void write(final Kryo kryo, final Output output, final Pair pair) {
+ kryo.writeClassAndObject(output, pair.getValue0());
+ kryo.writeClassAndObject(output, pair.getValue1());
+ }
+
+ @Override
+ public Pair read(final Kryo kryo, final Input input, final Class<Pair> pairClass) {
+ return Pair.with(kryo.readClassAndObject(input), kryo.readClassAndObject(input));
+ }
+}