You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2019/05/30 13:30:39 UTC
[tinkerpop] 01/01: TINKERPOP-2231 Removed deprecated
BulkLoader/DumperVertexProgram
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch TINKERPOP-2231
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit a9309099002b3e399465dc2eedd1852a1a00fbb0
Author: Stephen Mallette <sp...@genoprime.com>
AuthorDate: Thu May 30 09:28:18 2019 -0400
TINKERPOP-2231 Removed deprecated BulkLoader/DumperVertexProgram
Note that there are still some references to these classes in documentation. They were technically missed for removal in earlier versions. Will remove them from those branches for future releases and then merge forward.
---
CHANGELOG.asciidoc | 1 +
docs/src/upgrade/release-3.5.x.asciidoc | 8 +-
.../tinkerpop/gremlin/jsr223/CoreImports.java | 10 -
.../bulkdumping/BulkDumperVertexProgram.java | 101 -----
.../process/computer/bulkloading/BulkLoader.java | 144 ------
.../bulkloading/BulkLoaderVertexProgram.java | 499 ---------------------
.../bulkloading/IncrementalBulkLoader.java | 154 -------
.../computer/bulkloading/OneTimeBulkLoader.java | 111 -----
.../gremlin/process/ProcessComputerSuite.java | 4 -
.../bulkdumping/BulkDumperVertexProgramTest.java | 98 ----
.../bulkloading/BulkLoaderVertexProgramTest.java | 214 ---------
.../tinkerpop/gremlin/structure/RemoteGraph.java | 8 -
.../io/PersistedInputOutputRDDIntegrateTest.java | 84 ----
13 files changed, 8 insertions(+), 1428 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index fa9fdfb..c34d037 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
This release also includes changes from <<release-3-4-3, 3.4.3>>.
* Removed previously deprecated `TraversalSource.withRemote()`.
+* Removed previously deprecated `BulkDumperVertexProgram` and `BulkLoaderVertexProgram`.
== TinkerPop 3.4.0 (Avant-Gremlin Construction #3 for Theremin and Flowers)
diff --git a/docs/src/upgrade/release-3.5.x.asciidoc b/docs/src/upgrade/release-3.5.x.asciidoc
index 07a4c07..a394d35 100644
--- a/docs/src/upgrade/release-3.5.x.asciidoc
+++ b/docs/src/upgrade/release-3.5.x.asciidoc
@@ -34,7 +34,13 @@ Please see the link:https://github.com/apache/tinkerpop/blob/3.5.0/CHANGELOG.asc
The following deprecated classes, methods or fields have been removed in this version:
* `gremlin-core`
+** `org.apache.tinkerpop.gremlin.process.traversal.computer.bulkdumping.BulkDumperVertexProgram`
+** `org.apache.tinkerpop.gremlin.process.traversal.computer.bulkloading.BulkLoader`
+** `org.apache.tinkerpop.gremlin.process.traversal.computer.bulkloading.BulkLoaderVertexProgram`
+** `org.apache.tinkerpop.gremlin.process.traversal.computer.bulkloading.IncrementalBulkLoader`
+** `org.apache.tinkerpop.gremlin.process.traversal.computer.bulkloading.OneTimeBulkLoader`
** `org.apache.tinkerpop.gremlin.process.traversal.TraversalSource#withRemote(*)`
** `org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource#withRemote(*)`
-See: link:https://issues.apache.org/jira/browse/TINKERPOP-2080[TINKERPOP-2080]
\ No newline at end of file
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-2080[TINKERPOP-2080],
+link:https://issues.apache.org/jira/browse/TINKERPOP-2231[TINKERPOP-2231]
\ No newline at end of file
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/CoreImports.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/CoreImports.java
index cc7881c..48c8693 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/CoreImports.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/CoreImports.java
@@ -36,11 +36,6 @@ import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.bulkdumping.BulkDumperVertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.bulkloading.BulkLoader;
-import org.apache.tinkerpop.gremlin.process.computer.bulkloading.BulkLoaderVertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.bulkloading.IncrementalBulkLoader;
-import org.apache.tinkerpop.gremlin.process.computer.bulkloading.OneTimeBulkLoader;
import org.apache.tinkerpop.gremlin.process.computer.clone.CloneVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.clustering.connected.ConnectedComponentVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.clustering.peerpressure.ClusterCountMapReduce;
@@ -275,11 +270,6 @@ public final class CoreImports {
CLASS_IMPORTS.add(Memory.class);
CLASS_IMPORTS.add(VertexProgram.class);
CLASS_IMPORTS.add(CloneVertexProgram.class);
- CLASS_IMPORTS.add(BulkDumperVertexProgram.class);
- CLASS_IMPORTS.add(BulkLoader.class);
- CLASS_IMPORTS.add(BulkLoaderVertexProgram.class);
- CLASS_IMPORTS.add(IncrementalBulkLoader.class);
- CLASS_IMPORTS.add(OneTimeBulkLoader.class);
CLASS_IMPORTS.add(ClusterCountMapReduce.class);
CLASS_IMPORTS.add(ClusterPopulationMapReduce.class);
CLASS_IMPORTS.add(MemoryTraversalSideEffects.class);
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkDumperVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkDumperVertexProgram.java
deleted file mode 100644
index f70fcc2..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkDumperVertexProgram.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * 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.bulkdumping;
-
-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.clone.CloneVertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
-import org.javatuples.Tuple;
-
-import java.util.Collections;
-import java.util.Set;
-
-/**
- * @author Daniel Kuppitz (http://gremlin.guru)
- * @deprecated As of release 3.2.10, replaced by {@link CloneVertexProgram}.
- */
-@Deprecated
-public class BulkDumperVertexProgram implements VertexProgram<Tuple> {
-
- private BulkDumperVertexProgram() {
- }
-
- @Override
- public void setup(final Memory memory) {
- }
-
- @Override
- public void execute(final Vertex sourceVertex, final Messenger<Tuple> messenger, final Memory memory) {
- }
-
- @Override
- public boolean terminate(final Memory memory) {
- return true;
- }
-
- @Override
- public Set<MessageScope> getMessageScopes(final Memory memory) {
- return Collections.emptySet();
- }
-
- @SuppressWarnings({"CloneDoesntDeclareCloneNotSupportedException", "CloneDoesntCallSuperClone"})
- @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(BulkDumperVertexProgram.class);
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public BulkDumperVertexProgram create(final Graph graph) {
- return (BulkDumperVertexProgram) VertexProgram.createVertexProgram(graph, configuration);
- }
- }
-}
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
deleted file mode 100644
index 3767f73..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoader.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * 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)
- * @deprecated As of release 3.2.10, not directly replaced - consider graph provider specific bulk loading methods
- */
-@Deprecated
-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);
-
- /**
- * 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 default Edge createEdge(final Edge edge, final Vertex outVertex, final Vertex inVertex, final Graph graph, final GraphTraversalSource g) {
- final Edge result = outVertex.addEdge(edge.label(), inVertex);
- edge.properties().forEachRemaining(property -> result.property(property.key(), property.value()));
- return result;
- }
-
- /**
- * 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);
-
- /**
- * 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 default VertexProperty createVertexProperty(final VertexProperty<?> property, final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
- final VertexProperty result = vertex.property(property.key(), property.value());
- property.properties().forEachRemaining(metaProperty -> result.property(metaProperty.key(), metaProperty.value()));
- return result;
- }
-
- /**
- * 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 Vertex getVertex(final Vertex vertex, final Graph graph, final GraphTraversalSource g);
-
- /**
- * 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 g.V().hasId(id).next();
- }
-
- /**
- * @return Whether to use user supplied identifiers or not.
- */
- public boolean useUserSuppliedIds();
-
- /**
- * @return Whether to keep the original vertex identifiers in the target graph or not.
- */
- public boolean keepOriginalIds();
-
- /**
- * @return The name of the vertex property that is used to store the original vertex id.
- */
- public default String getVertexIdProperty() {
- return BulkLoaderVertexProgram.DEFAULT_BULK_LOADER_VERTEX_ID;
- }
-
- /**
- * Configures the BulkLoader instance.
- *
- * @param configuration The BulkLoader configuration.
- */
- public void configure(final Configuration configuration);
-}
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
deleted file mode 100644
index 3570422..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgram.java
+++ /dev/null
@@ -1,499 +0,0 @@
-/*
- * 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.commons.configuration.ConfigurationException;
-import org.apache.commons.configuration.ConfigurationUtils;
-import org.apache.commons.configuration.PropertiesConfiguration;
-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.VertexComputeKey;
-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.process.traversal.step.util.event.MutationListener;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.EventStrategy;
-import org.apache.tinkerpop.gremlin.structure.Direction;
-import org.apache.tinkerpop.gremlin.structure.Edge;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Property;
-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.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * @author Daniel Kuppitz (http://gremlin.guru)
- * @deprecated As of release 3.2.10, not directly replaced - consider graph provider specific bulk loading methods
- */
-@Deprecated
-public class BulkLoaderVertexProgram implements VertexProgram<Tuple> {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(BulkLoaderVertexProgram.class);
-
- public static final String BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX = "gremlin.bulkLoaderVertexProgram";
- public static final String BULK_LOADER_CLASS_CFG_KEY = String.join(".", BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX, "class");
- public static final String BULK_LOADER_VERTEX_ID_CFG_KEY = String.join(".", BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX, "vertexIdProperty");
- public static final String INTERMEDIATE_BATCH_SIZE_CFG_KEY = String.join(".", BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX, "intermediateBatchSize");
- public static final String KEEP_ORIGINAL_IDS_CFG_KEY = String.join(".", BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX, "keepOriginalIds");
- public static final String USER_SUPPLIED_IDS_CFG_KEY = String.join(".", BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX, "userSuppliedIds");
- public static final String WRITE_GRAPH_CFG_KEY = String.join(".", BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX, "writeGraph");
- public static final String DEFAULT_BULK_LOADER_VERTEX_ID = "bulkLoader.vertex.id";
-
- private final MessageScope messageScope;
- private final Set<VertexComputeKey> elementComputeKeys;
- private Configuration configuration;
- private BulkLoader bulkLoader;
- private Graph graph;
- private GraphTraversalSource g;
- private long intermediateBatchSize;
-
- private BulkLoadingListener listener;
-
- private BulkLoaderVertexProgram() {
- messageScope = MessageScope.Local.of(__::inE);
- elementComputeKeys = new HashSet<>();
- }
-
- private BulkLoader createBulkLoader() {
- final BulkLoader loader;
- final Configuration config = configuration.subset(BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX);
- if (config.containsKey("class")) {
- final String className = config.getString("class");
- try {
- final Class<?> bulkLoaderClass = Class.forName(className);
- loader = (BulkLoader) bulkLoaderClass.getConstructor().newInstance();
- } catch (ClassNotFoundException e) {
- LOGGER.error("Unable to find custom bulk loader class: {}", className);
- throw new IllegalStateException(e);
- } catch (Exception e) {
- LOGGER.error("Unable to create an instance of the given bulk loader class: {}", className);
- throw new IllegalStateException(e);
- }
- } else {
- loader = new IncrementalBulkLoader();
- }
- loader.configure(configuration);
- return loader;
- }
-
- /**
- * Eventually commits the current transaction and closes the current graph instance. commit() will be called
- * if close is set true, otherwise it will only be called if the intermediate batch size is set and reached.
- *
- * @param close Whether to close the current graph instance after calling commit() or not.
- */
- private void commit(final boolean close) {
- if (!close && (intermediateBatchSize == 0L || listener.mutations() < intermediateBatchSize))
- return;
- if (null != graph) {
- if (graph.features().graph().supportsTransactions()) {
- LOGGER.info("Committing transaction on Graph instance: {} [{} mutations]", graph, listener.mutations());
- try {
- graph.tx().commit();
- LOGGER.debug("Committed transaction on Graph instance: {}", graph);
- listener.resetCounter();
- } catch (Exception e) {
- LOGGER.error("Failed to commit transaction on Graph instance: {}", graph);
- graph.tx().rollback();
- listener.resetCounter();
- throw e;
- }
- }
- if (close) {
- 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 setup(final Memory memory) {
- }
-
- @Override
- public void loadState(final Graph graph, final Configuration config) {
- configuration = new BaseConfiguration();
- if (config != null) {
- ConfigurationUtils.copy(config, configuration);
- }
- intermediateBatchSize = configuration.getLong(INTERMEDIATE_BATCH_SIZE_CFG_KEY, 0L);
- elementComputeKeys.add(VertexComputeKey.of(DEFAULT_BULK_LOADER_VERTEX_ID, true));
- bulkLoader = createBulkLoader();
- }
-
- @Override
- public void storeState(final Configuration config) {
- VertexProgram.super.storeState(config);
- if (configuration != null) {
- ConfigurationUtils.copy(configuration, config);
- }
- }
-
- @Override
- public void workerIterationStart(final Memory memory) {
- if (null == graph) {
- graph = GraphFactory.open(configuration.subset(WRITE_GRAPH_CFG_KEY));
- LOGGER.info("Opened Graph instance: {}", graph);
- try {
- listener = new BulkLoadingListener();
- g = graph.traversal().withStrategies(EventStrategy.build().addListener(listener).create());
- } catch (Exception e) {
- try {
- graph.close();
- } catch (Exception e2) {
- LOGGER.warn("Failed to close Graph instance", e2);
- }
- throw e;
- }
- } else {
- LOGGER.warn("Leaked Graph instance: {}", graph);
- }
- }
-
- @Override
- public void workerIterationEnd(final Memory memory) {
- this.commit(true);
- }
-
- @Override
- public void execute(final Vertex sourceVertex, final Messenger<Tuple> messenger, final Memory memory) {
- try {
- executeInternal(sourceVertex, messenger, memory);
- } catch (Exception e) {
- if (graph.features().graph().supportsTransactions()) {
- graph.tx().rollback();
- }
- throw e;
- }
- }
-
- private void executeInternal(final Vertex sourceVertex, final Messenger<Tuple> messenger, final Memory memory) {
- if (memory.isInitialIteration()) {
- this.listener.resetStats();
- // 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();
- if (this.listener.isNewVertex()) {
- vpi.forEachRemaining(vp -> bulkLoader.createVertexProperty(vp, targetVertex, graph, g));
- } else {
- vpi.forEachRemaining(vp -> bulkLoader.getOrCreateVertexProperty(vp, targetVertex, graph, g));
- }
- this.commit(false);
- if (!bulkLoader.useUserSuppliedIds()) {
- // create an id pair and send it to all the vertex's incoming adjacent vertices
- sourceVertex.property(DEFAULT_BULK_LOADER_VERTEX_ID, targetVertex.id());
- messenger.sendMessage(messageScope, Pair.with(sourceVertex.id(), targetVertex.id()));
- }
- } else if (memory.getIteration() == 1) {
- if (bulkLoader.useUserSuppliedIds()) {
- final Vertex outV = bulkLoader.getVertex(sourceVertex, graph, g);
- final boolean incremental = outV.edges(Direction.OUT).hasNext();
- sourceVertex.edges(Direction.OUT).forEachRemaining(edge -> {
- final Vertex inV = bulkLoader.getVertex(edge.inVertex(), graph, g);
- if (incremental) {
- bulkLoader.getOrCreateEdge(edge, outV, inV, graph, g);
- } else {
- bulkLoader.createEdge(edge, outV, inV, graph, g);
- }
- this.commit(false);
- });
- } 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 with given the dummy id property
- final Object outVId = sourceVertex.value(DEFAULT_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);
- this.commit(false);
- });
- }
- } else if (memory.getIteration() == 2) {
- final Object vertexId = sourceVertex.value(DEFAULT_BULK_LOADER_VERTEX_ID);
- bulkLoader.getVertexById(vertexId, graph, g)
- .property(bulkLoader.getVertexIdProperty()).remove();
- this.commit(false);
- }
- }
-
- @Override
- public boolean terminate(final Memory memory) {
- switch (memory.getIteration()) {
- case 1:
- return bulkLoader.keepOriginalIds() || bulkLoader.getVertexIdProperty() == null;
- case 2:
- return true;
- }
- return false;
- }
-
- @Override
- public Set<VertexComputeKey> getVertexComputeKeys() {
- return elementComputeKeys;
- }
-
- @Override
- public Set<MessageScope> getMessageScopes(final Memory memory) {
- return Collections.singleton(messageScope);
- }
-
- @SuppressWarnings({"CloneDoesntDeclareCloneNotSupportedException", "CloneDoesntCallSuperClone"})
- @Override
- public VertexProgram<Tuple> clone() {
- return this;
- }
-
- @Override
- public GraphComputer.ResultGraph getPreferredResultGraph() {
- return GraphComputer.ResultGraph.ORIGINAL;
- }
-
- @Override
- public GraphComputer.Persist getPreferredPersist() {
- return GraphComputer.Persist.NOTHING;
- }
-
- @Override
- public String toString() {
- final StringBuilder sb = new StringBuilder();
- if (bulkLoader != null) {
- sb.append("bulkLoader=").append(bulkLoader.getClass().getSimpleName()).append(", ");
- sb.append("vertexIdProperty=").append(bulkLoader.getVertexIdProperty()).append(", ");
- sb.append("userSuppliedIds=").append(bulkLoader.useUserSuppliedIds()).append(", ");
- sb.append("keepOriginalIds=").append(bulkLoader.keepOriginalIds()).append(", ");
- } else {
- sb.append("bulkLoader=").append(bulkLoader).append(", ");
- }
- sb.append("batchSize=").append(intermediateBatchSize);
- return StringFactory.vertexProgramString(this, sb.toString());
- }
-
- public static Builder build() {
- return new Builder();
- }
-
- public static class Builder extends AbstractVertexProgramBuilder<Builder> {
-
- private Builder() {
- super(BulkLoaderVertexProgram.class);
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public BulkLoaderVertexProgram create(final Graph graph) {
- ConfigurationUtils.append(graph.configuration().subset(BULK_LOADER_VERTEX_PROGRAM_CFG_PREFIX), configuration);
- return (BulkLoaderVertexProgram) VertexProgram.createVertexProgram(graph, configuration);
- }
-
- private void setGraphConfigurationProperty(final String key, final Object value) {
- configuration.setProperty(String.join(".", WRITE_GRAPH_CFG_KEY, key), value);
- }
-
- /**
- * Sets the class name of the BulkLoader implementation to be used.
- */
- public Builder bulkLoader(final String className) {
- configuration.setProperty(BULK_LOADER_CLASS_CFG_KEY, className);
- return this;
- }
-
- /**
- * Sets the class of the BulkLoader implementation to be used.
- */
- public Builder bulkLoader(final Class<? extends BulkLoader> clazz) {
- return bulkLoader(clazz.getCanonicalName());
- }
-
- /**
- * Sets the name of the property that is used to store the original vertex identifiers in the target graph.
- */
- public Builder vertexIdProperty(final String name) {
- configuration.setProperty(BULK_LOADER_VERTEX_ID_CFG_KEY, name);
- return this;
- }
-
- /**
- * Specifies whether user supplied identifiers should be used when the bulk loader creates vertices in the
- * target graph.
- */
- public Builder userSuppliedIds(final boolean useUserSuppliedIds) {
- configuration.setProperty(USER_SUPPLIED_IDS_CFG_KEY, useUserSuppliedIds);
- return this;
- }
-
- /**
- * Specifies whether the original vertex identifiers should be kept in the target graph or not. In case of false
- * BulkLoaderVertexProgram will add another iteration to remove the properties and it won't be possible to use
- * the data for further incremental bulk loads.
- */
- public Builder keepOriginalIds(final boolean keepOriginalIds) {
- configuration.setProperty(KEEP_ORIGINAL_IDS_CFG_KEY, keepOriginalIds);
- return this;
- }
-
- /**
- * The batch size for a single transaction (number of vertices in the vertex loading stage; number of edges in
- * the edge loading stage).
- */
- public Builder intermediateBatchSize(final int batchSize) {
- configuration.setProperty(INTERMEDIATE_BATCH_SIZE_CFG_KEY, batchSize);
- return this;
- }
-
- /**
- * A configuration for the target graph that can be passed to GraphFactory.open().
- */
- public Builder writeGraph(final String configurationFile) throws ConfigurationException {
- return writeGraph(new PropertiesConfiguration(configurationFile));
- }
-
- /**
- * A configuration for the target graph that can be passed to GraphFactory.open().
- */
- public Builder writeGraph(final Configuration configuration) {
- configuration.getKeys().forEachRemaining(key -> setGraphConfigurationProperty(key, configuration.getProperty(key)));
- return this;
- }
- }
-
- @Override
- public Features getFeatures() {
- return new Features() {
- @Override
- public boolean requiresLocalMessageScopes() {
- return true;
- }
-
- @Override
- public boolean requiresVertexPropertyAddition() {
- return true;
- }
- };
- }
-
- static class BulkLoadingListener implements MutationListener {
-
- private long counter;
- private boolean isNewVertex;
-
- public BulkLoadingListener() {
- this.counter = 0L;
- this.isNewVertex = false;
- }
-
- public boolean isNewVertex() {
- return this.isNewVertex;
- }
-
- public long mutations() {
- return this.counter;
- }
-
- public void resetStats() {
- this.isNewVertex = false;
- }
-
- public void resetCounter() {
- this.counter = 0L;
- }
-
- @Override
- public void vertexAdded(final Vertex vertex) {
- this.isNewVertex = true;
- this.counter++;
- }
-
- @Override
- public void vertexRemoved(final Vertex vertex) {
- this.counter++;
- }
-
- @Override
- public void vertexPropertyChanged(final Vertex element, final VertexProperty oldValue, final Object setValue,
- final Object... vertexPropertyKeyValues) {
- this.counter++;
- }
-
- @Override
- public void vertexPropertyRemoved(final VertexProperty vertexProperty) {
- this.counter++;
- }
-
- @Override
- public void edgeAdded(final Edge edge) {
- this.counter++;
- }
-
- @Override
- public void edgeRemoved(final Edge edge) {
- this.counter++;
- }
-
- @Override
- public void edgePropertyChanged(final Edge element, final Property oldValue, final Object setValue) {
- this.counter++;
- }
-
- @Override
- public void edgePropertyRemoved(final Edge element, final Property property) {
- this.counter++;
- }
-
- @Override
- public void vertexPropertyPropertyChanged(final VertexProperty element, final Property oldValue, final Object setValue) {
- this.counter++;
- }
-
- @Override
- public void vertexPropertyPropertyRemoved(final VertexProperty element, final Property property) {
- this.counter++;
- }
- }
-}
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
deleted file mode 100644
index 5713773..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/IncrementalBulkLoader.java
+++ /dev/null
@@ -1,154 +0,0 @@
-/*
- * 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.Property;
-import org.apache.tinkerpop.gremlin.structure.T;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-
-import java.util.Iterator;
-
-/**
- * @author Daniel Kuppitz (http://gremlin.guru)
- * @deprecated As of release 3.2.10, not directly replaced - consider graph provider specific bulk loading methods
- */
-@Deprecated
-public class IncrementalBulkLoader implements BulkLoader {
-
- private String bulkLoaderVertexId = BulkLoaderVertexProgram.DEFAULT_BULK_LOADER_VERTEX_ID;
- private boolean keepOriginalIds = true;
- private boolean userSuppliedIds = false;
-
- /**
- * {@inheritDoc}
- */
- @Override
- public Vertex getOrCreateVertex(final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
- final Iterator<Vertex> iterator = useUserSuppliedIds()
- ? g.V().hasId(vertex.id())
- : g.V().has(vertex.label(), bulkLoaderVertexId, vertex.id().toString());
- return iterator.hasNext()
- ? iterator.next()
- : useUserSuppliedIds()
- ? g.addV(vertex.label()).property(T.id, vertex.id()).next()
- : g.addV(vertex.label()).property(bulkLoaderVertexId, vertex.id().toString()).next();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public Edge getOrCreateEdge(final Edge edge, final Vertex outVertex, final Vertex inVertex, final Graph graph, final GraphTraversalSource g) {
- final Edge e;
- final Traversal<Vertex, Edge> t = g.V(outVertex).outE(edge.label()).filter(__.inV().is(inVertex));
- if (t.hasNext()) {
- e = t.next();
- edge.properties().forEachRemaining(property -> {
- final Property<?> existing = e.property(property.key());
- if (!existing.isPresent() || !existing.value().equals(property.value())) {
- e.property(property.key(), property.value());
- }
- });
- } else {
- e = createEdge(edge, outVertex, inVertex, graph, g);
- }
- return e;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public VertexProperty getOrCreateVertexProperty(final VertexProperty<?> property, final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
- final VertexProperty<?> vp;
- final VertexProperty<?> existing = vertex.property(property.key());
- if (!existing.isPresent()) {
- return createVertexProperty(property, vertex, graph, g);
- }
- if (!existing.value().equals(property.value())) {
- vp = vertex.property(property.key(), property.value());
- } else {
- vp = existing;
- }
- property.properties().forEachRemaining(metaProperty -> {
- final Property<?> existing2 = vp.property(metaProperty.key());
- if (!existing2.isPresent() || !existing2.value().equals(metaProperty.value())) {
- vp.property(metaProperty.key(), metaProperty.value());
- }
- });
- return vp;
- }
-
- /**
- * {@inheritDoc}
- */
- @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(), bulkLoaderVertexId, vertex.id().toString()).next();
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public boolean useUserSuppliedIds() {
- return userSuppliedIds;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public boolean keepOriginalIds() {
- return keepOriginalIds;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public String getVertexIdProperty() {
- return bulkLoaderVertexId;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void configure(final Configuration configuration) {
- if (configuration.containsKey(BulkLoaderVertexProgram.BULK_LOADER_VERTEX_ID_CFG_KEY)) {
- bulkLoaderVertexId = configuration.getString(BulkLoaderVertexProgram.BULK_LOADER_VERTEX_ID_CFG_KEY);
- }
- if (configuration.containsKey(BulkLoaderVertexProgram.USER_SUPPLIED_IDS_CFG_KEY)) {
- userSuppliedIds = configuration.getBoolean(BulkLoaderVertexProgram.USER_SUPPLIED_IDS_CFG_KEY);
- }
- if (configuration.containsKey(BulkLoaderVertexProgram.KEEP_ORIGINAL_IDS_CFG_KEY)) {
- keepOriginalIds = configuration.getBoolean(BulkLoaderVertexProgram.KEEP_ORIGINAL_IDS_CFG_KEY);
- }
- }
-}
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/OneTimeBulkLoader.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/OneTimeBulkLoader.java
deleted file mode 100644
index ef34fce..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/OneTimeBulkLoader.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * 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.GraphTraversal;
-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;
-
-/**
- * {@link org.apache.tinkerpop.gremlin.process.computer.bulkloading.OneTimeBulkLoader} is a
- * {@link org.apache.tinkerpop.gremlin.process.computer.bulkloading.BulkLoader} implementation that should be used
- * for initial bulk loads. In contrast to {@link org.apache.tinkerpop.gremlin.process.computer.bulkloading.IncrementalBulkLoader}
- * it doesn't store temporary identifiers in the write graph nor does it attempt to find existing elements, instead it
- * only clones each element from the source graph.
- *
- * @author Daniel Kuppitz (http://gremlin.guru)
- * @deprecated As of release 3.2.10, not directly replaced - consider graph provider specific bulk loading methods
- */
-@Deprecated
-public class OneTimeBulkLoader implements BulkLoader {
-
- private boolean userSuppliedIds = false;
-
- /**
- * Creates a clone of the given vertex in the given graph.
- */
- @Override
- public Vertex getOrCreateVertex(final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
- final GraphTraversal<Vertex, Vertex> t = g.addV(vertex.label());
- return (useUserSuppliedIds() ? t.property(T.id, vertex.id()) : t).next();
- }
-
- /**
- * Creates a clone of the given edge between the given in- and out-vertices.
- */
- @Override
- public Edge getOrCreateEdge(final Edge edge, final Vertex outVertex, final Vertex inVertex, final Graph graph, final GraphTraversalSource g) {
- return createEdge(edge, outVertex, inVertex, graph, g);
- }
-
- /**
- * Creates a clone of the given property for the given vertex.
- */
- @Override
- public VertexProperty getOrCreateVertexProperty(final VertexProperty<?> property, final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
- return createVertexProperty(property, vertex, graph, g);
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public Vertex getVertex(final Vertex vertex, final Graph graph, final GraphTraversalSource g) {
- return getVertexById(vertex.id(), graph, g);
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public boolean useUserSuppliedIds() {
- return userSuppliedIds;
- }
-
- /**
- * Always returns false.
- */
- @Override
- public boolean keepOriginalIds() {
- return false;
- }
-
- /**
- * Always returns null.
- */
- @Override
- public String getVertexIdProperty() {
- return null;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void configure(final Configuration configuration) {
- if (configuration.containsKey(BulkLoaderVertexProgram.USER_SUPPLIED_IDS_CFG_KEY)) {
- userSuppliedIds = configuration.getBoolean(BulkLoaderVertexProgram.USER_SUPPLIED_IDS_CFG_KEY);
- }
- }
-}
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
index 0951b86..17d1f13 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
@@ -21,8 +21,6 @@ package org.apache.tinkerpop.gremlin.process;
import org.apache.tinkerpop.gremlin.AbstractGremlinSuite;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputerTest;
-import org.apache.tinkerpop.gremlin.process.computer.bulkdumping.BulkDumperVertexProgramTest;
-import org.apache.tinkerpop.gremlin.process.computer.bulkloading.BulkLoaderVertexProgramTest;
import org.apache.tinkerpop.gremlin.process.computer.clone.CloneVertexProgramTest;
import org.apache.tinkerpop.gremlin.process.computer.clustering.peerpressure.PeerPressureVertexProgramTest;
import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgramTest;
@@ -201,8 +199,6 @@ public class ProcessComputerSuite extends AbstractGremlinSuite {
PageRankVertexProgramTest.class,
PeerPressureVertexProgramTest.class,
ShortestPathVertexProgramTest.class,
- BulkLoaderVertexProgramTest.class,
- BulkDumperVertexProgramTest.class,
CloneVertexProgramTest.class,
// creations
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkDumperVertexProgramTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkDumperVertexProgramTest.java
deleted file mode 100644
index 62104c0..0000000
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkDumperVertexProgramTest.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * 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.bulkdumping;
-
-import org.apache.tinkerpop.gremlin.LoadGraphWith;
-import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest;
-import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
-import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
-import org.apache.tinkerpop.gremlin.structure.Direction;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-import org.junit.Test;
-
-import static org.apache.tinkerpop.gremlin.LoadGraphWith.GraphData.MODERN;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-/**
- * @author Daniel Kuppitz (http://gremlin.guru)
- */
-public class BulkDumperVertexProgramTest extends AbstractGremlinProcessTest {
-
- @Test
- @LoadGraphWith(MODERN)
- public void shouldDumpWholeGraph() throws Exception {
- if (graphProvider.getGraphComputer(graph).features().supportsResultGraphPersistCombination(GraphComputer.ResultGraph.NEW, GraphComputer.Persist.EDGES)) {
- final ComputerResult result = graphProvider.getGraphComputer(graph).program(BulkDumperVertexProgram.build().create(graph)).submit().get();
- result.graph().traversal().V().forEachRemaining(v -> {
- assertEquals(2, v.keys().size());
- assertTrue(v.keys().contains("name"));
- assertTrue(v.keys().contains("age") || v.keys().contains("lang"));
- assertEquals(1, IteratorUtils.count(v.values("name")));
- assertEquals(1, IteratorUtils.count(v.values("age", "lang")));
- final String name = v.value("name");
- if (name.equals("marko")) {
- assertEquals("person", v.label());
- assertEquals(Integer.valueOf(29), v.value("age"));
- assertEquals(3, IteratorUtils.count(v.edges(Direction.OUT)));
- assertEquals(2, IteratorUtils.count(v.edges(Direction.OUT, "knows")));
- assertEquals(1, IteratorUtils.count(v.edges(Direction.OUT, "created")));
- assertEquals(0, IteratorUtils.count(v.edges(Direction.IN)));
- } else if (name.equals("vadas")) {
- assertEquals("person", v.label());
- assertEquals(Integer.valueOf(27), v.value("age"));
- assertEquals(0, IteratorUtils.count(v.edges(Direction.OUT)));
- assertEquals(1, IteratorUtils.count(v.edges(Direction.IN)));
- assertEquals(1, IteratorUtils.count(v.edges(Direction.IN, "knows")));
- } else if (name.equals("lop")) {
- assertEquals("software", v.label());
- assertEquals("java", v.value("lang"));
- assertEquals(0, IteratorUtils.count(v.edges(Direction.OUT)));
- assertEquals(3, IteratorUtils.count(v.edges(Direction.IN)));
- assertEquals(3, IteratorUtils.count(v.edges(Direction.IN, "created")));
- } else if (name.equals("josh")) {
- assertEquals("person", v.label());
- assertEquals(Integer.valueOf(32), v.value("age"));
- assertEquals(2, IteratorUtils.count(v.edges(Direction.OUT)));
- assertEquals(2, IteratorUtils.count(v.edges(Direction.OUT, "created")));
- assertEquals(1, IteratorUtils.count(v.edges(Direction.IN)));
- assertEquals(1, IteratorUtils.count(v.edges(Direction.IN, "knows")));
- } else if (name.equals("ripple")) {
- assertEquals("software", v.label());
- assertEquals("java", v.value("lang"));
- assertEquals(0, IteratorUtils.count(v.edges(Direction.OUT)));
- assertEquals(1, IteratorUtils.count(v.edges(Direction.IN)));
- assertEquals(1, IteratorUtils.count(v.edges(Direction.IN, "created")));
- } else if (name.equals("peter")) {
- assertEquals("person", v.label());
- assertEquals(Integer.valueOf(35), v.value("age"));
- assertEquals(1, IteratorUtils.count(v.edges(Direction.OUT)));
- assertEquals(1, IteratorUtils.count(v.edges(Direction.OUT, "created")));
- assertEquals(0, IteratorUtils.count(v.edges(Direction.IN)));
- } else
- throw new IllegalStateException("The following vertex should not exist in the graph: " + name);
- });
- assertEquals(3.5, (Double) result.graph().traversal().E().values("weight").sum().next(), 0.01);
- assertEquals(1.5, (Double) result.graph().traversal().E().hasLabel("knows").values("weight").sum().next(), 0.01);
- assertEquals(2.0, (Double) result.graph().traversal().E().hasLabel("created").values("weight").sum().next(), 0.01);
- assertEquals(result.memory().getIteration(), 0);
- assertEquals(result.memory().asMap().size(), 0);
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgramTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgramTest.java
deleted file mode 100644
index d6db046..0000000
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgramTest.java
+++ /dev/null
@@ -1,214 +0,0 @@
-/*
- * 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.LoadGraphWith;
-import org.apache.tinkerpop.gremlin.TestHelper;
-import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest;
-import org.apache.tinkerpop.gremlin.process.IgnoreEngine;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalEngine;
-import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
-import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
-import org.apache.tinkerpop.gremlin.structure.Direction;
-import org.apache.tinkerpop.gremlin.structure.Element;
-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.util.iterator.IteratorUtils;
-import org.junit.After;
-import org.junit.Test;
-
-import java.io.File;
-import java.lang.reflect.Field;
-import java.util.Iterator;
-import java.util.function.Function;
-
-import static org.apache.tinkerpop.gremlin.LoadGraphWith.GraphData.MODERN;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-/**
- * @author Daniel Kuppitz (http://gremlin.guru)
- */
-public class BulkLoaderVertexProgramTest extends AbstractGremlinProcessTest {
-
- final static String TINKERGRAPH_LOCATION = TestHelper.makeTestDataDirectory(BulkLoaderVertexProgramTest.class) + "tinkertest.kryo";
-
- private BulkLoader getBulkLoader(final BulkLoaderVertexProgram blvp) throws Exception {
- final Field field = BulkLoaderVertexProgram.class.getDeclaredField("bulkLoader");
- field.setAccessible(true);
- return (BulkLoader) field.get(blvp);
- }
-
- private Configuration getWriteGraphConfiguration() {
- final Configuration configuration = new BaseConfiguration();
- configuration.setProperty(Graph.GRAPH, "org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph");
- configuration.setProperty("gremlin.tinkergraph.graphLocation", TINKERGRAPH_LOCATION);
- configuration.setProperty("gremlin.tinkergraph.graphFormat", "gryo");
- return configuration;
- }
-
- private Graph getWriteGraph() {
- return GraphFactory.open(getWriteGraphConfiguration());
- }
-
- @After
- public void cleanup() {
- final File graph = new File(TINKERGRAPH_LOCATION);
- assertTrue(!graph.exists() || graph.delete());
- }
-
- @Test
- public void shouldUseIncrementalBulkLoaderByDefault() throws Exception {
- final BulkLoader loader = getBulkLoader(BulkLoaderVertexProgram.build().create(graph));
- assertTrue(loader instanceof IncrementalBulkLoader);
- assertTrue(loader.keepOriginalIds());
- assertFalse(loader.useUserSuppliedIds());
- }
-
- @Test
- @LoadGraphWith(MODERN)
- public void shouldStoreOriginalIds() throws Exception {
- final BulkLoaderVertexProgram blvp = BulkLoaderVertexProgram.build()
- .userSuppliedIds(false)
- .writeGraph(getWriteGraphConfiguration()).create(graph);
- final BulkLoader loader = getBulkLoader(blvp);
- assertFalse(loader.useUserSuppliedIds());
- graphProvider.getGraphComputer(graph).workers(1).program(blvp).submit().get();
- assertGraphEquality(graph, getWriteGraph(), v -> v.value(loader.getVertexIdProperty()));
- }
-
- @Test
- @LoadGraphWith(MODERN)
- public void shouldNotStoreOriginalIds() throws Exception {
- final BulkLoaderVertexProgram blvp = BulkLoaderVertexProgram.build()
- .userSuppliedIds(true)
- .writeGraph(getWriteGraphConfiguration()).create(graph);
- final BulkLoader loader = getBulkLoader(blvp);
- assertTrue(loader.useUserSuppliedIds());
- graphProvider.getGraphComputer(graph).workers(1).program(blvp).submit().get();
- assertGraphEquality(graph, getWriteGraph());
- }
-
- @Test
- @LoadGraphWith(MODERN)
- public void shouldOverwriteExistingElements() throws Exception {
- final BulkLoaderVertexProgram blvp = BulkLoaderVertexProgram.build()
- .userSuppliedIds(true)
- .writeGraph(getWriteGraphConfiguration()).create(graph);
- graphProvider.getGraphComputer(graph).workers(1).program(blvp).submit().get(); // initial
- graphProvider.getGraphComputer(graph).workers(1).program(blvp).submit().get(); // incremental
- assertGraphEquality(graph, getWriteGraph());
- }
-
- @Test
- @LoadGraphWith(MODERN)
- @IgnoreEngine(TraversalEngine.Type.COMPUTER) // we can't modify the graph in computer mode
- public void shouldProperlyHandleMetaProperties() throws Exception {
- graph.traversal().V().has("name", "marko").properties("name").property("alias", "okram").iterate();
- final BulkLoaderVertexProgram blvp = BulkLoaderVertexProgram.build()
- .userSuppliedIds(true)
- .writeGraph(getWriteGraphConfiguration()).create(graph);
- graphProvider.getGraphComputer(graph).workers(1).program(blvp).submit().get();
- assertGraphEquality(graph, getWriteGraph());
- }
-
- @Test
- @LoadGraphWith(MODERN)
- public void shouldUseOneTimeBulkLoader() throws Exception {
- for (int iteration = 1; iteration <= 2; iteration++) {
- final BulkLoaderVertexProgram blvp = BulkLoaderVertexProgram.build()
- .bulkLoader(OneTimeBulkLoader.class)
- .writeGraph(getWriteGraphConfiguration()).create(graph);
- final BulkLoader loader = getBulkLoader(blvp);
- assertTrue(loader instanceof OneTimeBulkLoader);
- graphProvider.getGraphComputer(graph).workers(1).program(blvp).submit().get();
- final Graph result = getWriteGraph();
- assertEquals(6 * iteration, IteratorUtils.count(result.vertices()));
- assertEquals(6 * iteration, IteratorUtils.count(result.edges()));
- result.close();
- }
- }
-
- @Test
- @LoadGraphWith(MODERN)
- public void shouldUseOneTimeBulkLoaderWithUserSuppliedIds() throws Exception {
- final BulkLoaderVertexProgram blvp = BulkLoaderVertexProgram.build()
- .bulkLoader(OneTimeBulkLoader.class)
- .userSuppliedIds(true)
- .writeGraph(getWriteGraphConfiguration()).create(graph);
- final BulkLoader loader = getBulkLoader(blvp);
- assertTrue(loader instanceof OneTimeBulkLoader);
- graphProvider.getGraphComputer(graph).workers(1).program(blvp).submit().get();
- final Graph result = getWriteGraph();
- assertEquals(6, IteratorUtils.count(result.vertices()));
- assertEquals(6, IteratorUtils.count(result.edges()));
- result.close();
- }
-
- private static void assertGraphEquality(final Graph source, final Graph target) {
- assertGraphEquality(source, target, Element::id);
- }
-
- private static void assertGraphEquality(final Graph source, final Graph target, final Function<Vertex, Object> idAccessor) {
- final GraphTraversalSource tg = target.traversal();
- assertEquals(IteratorUtils.count(source.vertices()), IteratorUtils.count(target.vertices()));
- assertEquals(IteratorUtils.count(target.edges()), IteratorUtils.count(target.edges()));
- source.vertices().forEachRemaining(originalVertex -> {
- Vertex tmpVertex = null;
- final Iterator<Vertex> vertexIterator = target.vertices();
- while (vertexIterator.hasNext()) {
- final Vertex v = vertexIterator.next();
- if (idAccessor.apply(v).toString().equals(originalVertex.id().toString())) {
- tmpVertex = v;
- break;
- }
- }
- assertNotNull(tmpVertex);
- final Vertex clonedVertex = tmpVertex;
- assertEquals(IteratorUtils.count(originalVertex.edges(Direction.IN)), IteratorUtils.count(clonedVertex.edges(Direction.IN)));
- assertEquals(IteratorUtils.count(originalVertex.edges(Direction.OUT)), IteratorUtils.count(clonedVertex.edges(Direction.OUT)));
- assertEquals(originalVertex.label(), clonedVertex.label());
- originalVertex.properties().forEachRemaining(originalProperty -> {
- VertexProperty clonedProperty = null;
- final Iterator<VertexProperty<Object>> vertexPropertyIterator = clonedVertex.properties(originalProperty.key());
- while (vertexPropertyIterator.hasNext()) {
- final VertexProperty p = vertexPropertyIterator.next();
- if (p.value().equals(originalProperty.value())) {
- clonedProperty = p;
- break;
- }
- }
- assertNotNull(clonedProperty);
- assertEquals(originalProperty.isPresent(), clonedProperty.isPresent());
- assertEquals(originalProperty.value(), clonedProperty.value());
- });
- originalVertex.edges(Direction.OUT).forEachRemaining(originalEdge -> {
- GraphTraversal t = tg.V(clonedVertex).outE(originalEdge.label());
- originalEdge.properties().forEachRemaining(p -> t.has(p.key(), p.value()));
- assertTrue(t.hasNext());
- });
- });
- }
-}
\ No newline at end of file
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/structure/RemoteGraph.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/structure/RemoteGraph.java
index 954118b..b1af305 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/structure/RemoteGraph.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/structure/RemoteGraph.java
@@ -79,14 +79,6 @@ import java.util.Iterator;
method = "*",
reason = "RemoteGraph does not support direct Graph.compute() access")
@Graph.OptOut(
- test = "org.apache.tinkerpop.gremlin.process.computer.bulkloading.BulkLoaderVertexProgramTest",
- method = "*",
- reason = "RemoteGraph does not support direct Graph.compute() access")
-@Graph.OptOut(
- test = "org.apache.tinkerpop.gremlin.process.computer.bulkdumping.BulkDumperVertexProgramTest",
- method = "*",
- reason = "RemoteGraph does not support direct Graph.compute() access")
-@Graph.OptOut(
test = "org.apache.tinkerpop.gremlin.process.computer.clone.CloneVertexProgramTest",
method = "*",
reason = "RemoteGraph does not support direct Graph.compute() access")
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDIntegrateTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDIntegrateTest.java
index cc32a06..94e2eb4 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDIntegrateTest.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDIntegrateTest.java
@@ -19,7 +19,6 @@
package org.apache.tinkerpop.gremlin.spark.structure.io;
-import org.apache.commons.configuration.BaseConfiguration;
import org.apache.commons.configuration.Configuration;
import org.apache.spark.storage.StorageLevel;
import org.apache.tinkerpop.gremlin.TestHelper;
@@ -28,7 +27,6 @@ import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoOutputFormat;
import org.apache.tinkerpop.gremlin.process.computer.Computer;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
-import org.apache.tinkerpop.gremlin.process.computer.bulkloading.BulkLoaderVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankMapReduce;
import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
@@ -38,9 +36,7 @@ import org.apache.tinkerpop.gremlin.spark.process.computer.SparkGraphComputer;
import org.apache.tinkerpop.gremlin.spark.process.computer.SparkHadoopGraphProvider;
import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.io.IoCore;
import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
import org.junit.Test;
import java.util.Arrays;
@@ -208,86 +204,6 @@ public class PersistedInputOutputRDDIntegrateTest extends AbstractSparkTest {
}
@Test
- public void testBulkLoaderVertexProgramChain() throws Exception {
- Spark.create("local[4]");
- final String rddName = TestHelper.makeTestDataDirectory(PersistedInputOutputRDDIntegrateTest.class, UUID.randomUUID().toString());
- final Configuration readConfiguration = super.getBaseConfiguration();
- readConfiguration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
- readConfiguration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern-v3d0.kryo"));
- readConfiguration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
- readConfiguration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, rddName);
- readConfiguration.setProperty(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true);
- Graph pageRankGraph = GraphFactory.open(readConfiguration);
- ///////////////
- final Configuration writeConfiguration = new BaseConfiguration();
- writeConfiguration.setProperty(Graph.GRAPH, TinkerGraph.class.getCanonicalName());
- writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_FORMAT, "gryo");
- writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION, TestHelper.makeTestDataDirectory(PersistedInputOutputRDDIntegrateTest.class) + "testBulkLoaderVertexProgramChain.kryo");
- final Graph bulkLoaderGraph = pageRankGraph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.VERTEX_PROPERTIES).program(PageRankVertexProgram.build().create(pageRankGraph)).submit().get().graph();
- bulkLoaderGraph.compute(SparkGraphComputer.class)
- .persist(GraphComputer.Persist.NOTHING)
- .workers(1)
- .configure(Constants.GREMLIN_HADOOP_GRAPH_READER, PersistedInputRDD.class.getCanonicalName())
- .configure(Constants.GREMLIN_HADOOP_INPUT_LOCATION, rddName)
- .configure(Constants.GREMLIN_HADOOP_GRAPH_WRITER, null)
- .configure(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, null)
- .program(BulkLoaderVertexProgram.build().userSuppliedIds(true).writeGraph(writeConfiguration).create(bulkLoaderGraph))
- .submit().get();
- ////
- assertTrue(Spark.hasRDD(Constants.getGraphLocation(rddName)));
- assertEquals(1, Spark.getContext().getPersistentRDDs().size());
- ////
- final Graph graph = TinkerGraph.open();
- final GraphTraversalSource g = graph.traversal();
- graph.io(IoCore.gryo()).readGraph(TestHelper.makeTestDataDirectory(PersistedInputOutputRDDIntegrateTest.class) + "testBulkLoaderVertexProgramChain.kryo");
- assertEquals(6l, g.V().count().next().longValue());
- assertEquals(0l, g.E().count().next().longValue());
- assertEquals("marko", g.V().has("name", "marko").values("name").next());
- assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
- ////
- Spark.close();
- }
-
- @Test
- public void testBulkLoaderVertexProgramChainWithInputOutputHelperMapping() throws Exception {
- Spark.create("local[4]");
-
- final String rddName = TestHelper.makeTestDataDirectory(PersistedInputOutputRDDIntegrateTest.class, UUID.randomUUID().toString());
- final Configuration readConfiguration = super.getBaseConfiguration();
- readConfiguration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
- readConfiguration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern-v3d0.kryo"));
- readConfiguration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
- readConfiguration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, rddName);
- readConfiguration.setProperty(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true);
- Graph pageRankGraph = GraphFactory.open(readConfiguration);
- ///////////////
- final Configuration writeConfiguration = new BaseConfiguration();
- writeConfiguration.setProperty(Graph.GRAPH, TinkerGraph.class.getCanonicalName());
- writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_FORMAT, "gryo");
- writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION, TestHelper.makeTestDataDirectory(PersistedInputOutputRDDIntegrateTest.class) + "testBulkLoaderVertexProgramChainWithInputOutputHelperMapping.kryo");
- final Graph bulkLoaderGraph = pageRankGraph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.EDGES).program(PageRankVertexProgram.build().create(pageRankGraph)).submit().get().graph();
- bulkLoaderGraph.compute(SparkGraphComputer.class)
- .persist(GraphComputer.Persist.NOTHING)
- .workers(1)
- .program(BulkLoaderVertexProgram.build().userSuppliedIds(true).writeGraph(writeConfiguration).create(bulkLoaderGraph))
- .submit().get();
- ////
- Spark.create(readConfiguration);
- assertTrue(Spark.hasRDD(Constants.getGraphLocation(rddName)));
- assertEquals(1, Spark.getContext().getPersistentRDDs().size());
- ////
- final Graph graph = TinkerGraph.open();
- final GraphTraversalSource g = graph.traversal();
- graph.io(IoCore.gryo()).readGraph(TestHelper.makeTestDataDirectory(PersistedInputOutputRDDIntegrateTest.class) + "testBulkLoaderVertexProgramChainWithInputOutputHelperMapping.kryo");
- assertEquals(6l, g.V().count().next().longValue());
- assertEquals(6l, g.E().count().next().longValue());
- assertEquals("marko", g.V().has("name", "marko").values("name").next());
- assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
- ////
- Spark.close();
- }
-
- @Test
public void testComplexChain() throws Exception {
Spark.create("local[4]");