You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ok...@apache.org on 2016/02/12 19:57:14 UTC
incubator-tinkerpop git commit: Found a couple of bugs around job
chaining in all GraphComputer implementations. Wrote a solid
GraphComputerTest to ensure job chaining behaves as expected. Still a few
more things to handle -- the Traversal.pageRank() tes
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1140 66d86cdfb -> cd1a37946
Found a couple of bugs around job chaining in all GraphComputer implementations. Wrote a solid GraphComputerTest to ensure job chaining behaves as expected. Still a few more things to handle -- the Traversal.pageRank() tests in Giraph don't pass and I'm not sure why.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/cd1a3794
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/cd1a3794
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/cd1a3794
Branch: refs/heads/TINKERPOP-1140
Commit: cd1a37946848d28ad33d464fedac1655cadd5320
Parents: 66d86cd
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri Feb 12 11:57:05 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri Feb 12 11:57:05 2016 -0700
----------------------------------------------------------------------
.../process/computer/GiraphGraphComputer.java | 27 +++---
.../traversal/TraversalVertexProgram.java | 2 +-
.../mapreduce/TraverserMapReduce.java | 93 ++++++++++----------
.../process/traversal/util/TraversalHelper.java | 22 +++++
.../process/util/TraversalHelperTest.java | 22 +++--
.../process/GroovyProcessComputerSuite.java | 10 ---
.../process/computer/GraphComputerTest.java | 52 ++++++++++-
.../process/computer/util/MapReduceHelper.java | 25 +++---
.../process/computer/SparkGraphComputer.java | 3 +-
.../spark/structure/io/OutputFormatRDD.java | 9 +-
.../computer/SparkHadoopGraphProvider.java | 10 ++-
11 files changed, 182 insertions(+), 93 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
index bc6df7a..f031137 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Cluster;
import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
@@ -123,6 +124,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
private Future<ComputerResult> submitWithExecutor(final Executor exec) {
final long startTime = System.currentTimeMillis();
+ final Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(this.giraphConfiguration);
return CompletableFuture.<ComputerResult>supplyAsync(() -> {
try {
final FileSystem fs = FileSystem.get(this.giraphConfiguration);
@@ -133,7 +135,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
throw new IllegalStateException(e.getMessage(), e);
}
this.memory.setRuntime(System.currentTimeMillis() - startTime);
- return new DefaultComputerResult(InputOutputHelper.getOutputGraph(ConfUtil.makeApacheConfiguration(this.giraphConfiguration), this.resultGraph, this.persist), this.memory.asImmutable());
+ return new DefaultComputerResult(InputOutputHelper.getOutputGraph(apacheConfiguration, this.resultGraph, this.persist), this.memory.asImmutable());
}, exec);
}
@@ -157,6 +159,9 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
if (e.getCause() instanceof NumberFormatException)
throw new NotSerializableException("The provided traversal is not serializable and thus, can not be distributed across the cluster");
}
+ // remove historic combiners in configuration propagation (this occurs when job chaining)
+ if (!this.vertexProgram.getMessageCombiner().isPresent())
+ this.giraphConfiguration.unset(GiraphConstants.MESSAGE_COMBINER_CLASS.getKey());
// split required workers across system (open map slots + max threads per machine = total amount of TinkerPop workers)
if (!this.useWorkerThreadsInConfiguration) {
final Cluster cluster = new Cluster(GiraphGraphComputer.this.giraphConfiguration);
@@ -172,20 +177,22 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
this.giraphConfiguration.setNumComputeThreads(threadsPerMapper);
}
}
+ // prepare the giraph vertex-centric computing job
+ final GiraphJob job = new GiraphJob(this.giraphConfiguration, Constants.GREMLIN_HADOOP_GIRAPH_JOB_PREFIX + this.vertexProgram);
+ job.getInternalJob().setJarByClass(GiraphGraphComputer.class);
+ this.logger.info(Constants.GREMLIN_HADOOP_GIRAPH_JOB_PREFIX + this.vertexProgram);
// handle input paths (if any)
String inputLocation = this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION, null);
if (null != inputLocation && FileInputFormat.class.isAssignableFrom(this.giraphConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputFormat.class))) {
inputLocation = Constants.getSearchGraphLocation(inputLocation, storage).orElse(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION));
- apacheConfiguration.setProperty(Constants.MAPREDUCE_INPUT_FILEINPUTFORMAT_INPUTDIR, FileSystem.get(this.giraphConfiguration).getFileStatus(new Path(inputLocation)).getPath().toString());
- this.giraphConfiguration.set(Constants.MAPREDUCE_INPUT_FILEINPUTFORMAT_INPUTDIR, FileSystem.get(this.giraphConfiguration).getFileStatus(new Path(inputLocation)).getPath().toString());
+ FileInputFormat.setInputPaths(job.getInternalJob(), new Path(inputLocation));
+ }
+ // handle output paths (if any)
+ String outputLocation = this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, null);
+ if (null != outputLocation && FileOutputFormat.class.isAssignableFrom(this.giraphConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, OutputFormat.class))) {
+ outputLocation = Constants.getGraphLocation(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION));
+ FileOutputFormat.setOutputPath(job.getInternalJob(), new Path(outputLocation));
}
- // handle output paths
- final Path outputPath = new Path(Constants.getGraphLocation(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION)));
- // prepare the giraph vertex-centric computing job
- final GiraphJob job = new GiraphJob(this.giraphConfiguration, Constants.GREMLIN_HADOOP_GIRAPH_JOB_PREFIX + this.vertexProgram);
- FileOutputFormat.setOutputPath(job.getInternalJob(), outputPath);
- job.getInternalJob().setJarByClass(GiraphGraphComputer.class);
- this.logger.info(Constants.GREMLIN_HADOOP_GIRAPH_JOB_PREFIX + this.vertexProgram);
// execute the job and wait until it completes (if it fails, throw an exception)
if (!job.run(true))
throw new IllegalStateException("The GiraphGraphComputer job failed -- aborting all subsequent MapReduce jobs"); // how do I get the exception that occured?
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
index 573e826..cdd7362 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
@@ -121,7 +121,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
this.mapReducers.add(mapReducer.getMapReduce());
}
if (!(this.traversal.getEndStep() instanceof SideEffectCapStep) && !(this.traversal.getEndStep() instanceof ReducingBarrierStep))
- this.mapReducers.add(new TraverserMapReduce(this.traversal));
+ this.mapReducers.add(new TraverserMapReduce(this.traversal.clone()));
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java
index 10f8e9a..07294f9 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java
@@ -20,27 +20,22 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal.step.sideEffect.
import org.apache.commons.configuration.Configuration;
import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
+import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
+import org.apache.tinkerpop.gremlin.process.computer.util.VertexProgramHelper;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
-import org.apache.tinkerpop.gremlin.process.traversal.lambda.ElementValueTraversal;
-import org.apache.tinkerpop.gremlin.process.traversal.lambda.TokenTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.DedupGlobalStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.RangeGlobalStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.TailGlobalStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.EdgeVertexStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.OrderGlobalStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.PropertiesStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.PropertyMapStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.VertexStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.CollectingBarrierStep;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
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.util.Attachable;
import org.apache.tinkerpop.gremlin.util.function.ChainedComparator;
@@ -53,16 +48,18 @@ import java.util.Optional;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public final class TraverserMapReduce extends StaticMapReduce<Comparable, Traverser<?>, Comparable, Traverser<?>, Iterator<Traverser<?>>> {
+public final class TraverserMapReduce implements MapReduce<Comparable, Traverser<?>, Comparable, Traverser<?>, Iterator<Traverser<?>>> {
+
+ public static final String TRAVERSER_TRAVERSAL = "gremlin.traverserMapReduce.traversal";
public static final String TRAVERSERS = Graph.Hidden.hide("traversers");
private Traversal.Admin<?, ?> traversal;
- private Optional<Comparator<Comparable>> comparator = Optional.empty();
- private Optional<CollectingBarrierStep<?>> collectingBarrierStep = Optional.empty();
+ private Comparator<Comparable> comparator = null;
+ private CollectingBarrierStep<?> collectingBarrierStep = null;
private boolean attachHaltedTraverser = false;
- private Optional<RangeGlobalStep<?>> rangeGlobalStep = Optional.empty();
- private Optional<TailGlobalStep<?>> tailGlobalStep = Optional.empty();
+ private RangeGlobalStep<?> rangeGlobalStep = null;
+ private TailGlobalStep<?> tailGlobalStep = null;
private boolean dedupGlobal = false;
private TraverserMapReduce() {
@@ -74,23 +71,42 @@ public final class TraverserMapReduce extends StaticMapReduce<Comparable, Traver
}
@Override
+ public TraverserMapReduce clone() {
+ try {
+ final TraverserMapReduce clone = (TraverserMapReduce) super.clone();
+ clone.traversal = this.traversal.clone();
+ clone.genericLoadState();
+ return clone;
+ } catch (final CloneNotSupportedException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+
+ @Override
public void loadState(final Graph graph, final Configuration configuration) {
- this.traversal = TraversalVertexProgram.getTraversal(graph, configuration);
+ this.traversal = VertexProgramHelper.deserialize(configuration, TRAVERSER_TRAVERSAL);
this.genericLoadState();
}
+ @Override
+ public void storeState(final Configuration configuration) {
+ MapReduce.super.storeState(configuration);
+ VertexProgramHelper.serialize(this.traversal, configuration, TRAVERSER_TRAVERSAL);
+ }
+
private void genericLoadState() {
final Step<?, ?> traversalEndStep = traversal.getEndStep();
- this.comparator = Optional.ofNullable(traversalEndStep instanceof OrderGlobalStep ? new ChainedComparator<Comparable>(((OrderGlobalStep) traversalEndStep).getComparators()) : null);
if (traversalEndStep instanceof CollectingBarrierStep) {
- this.collectingBarrierStep = Optional.of((CollectingBarrierStep<?>) traversalEndStep); // why no clone() like the others?
- if (this.collectingBarrierStep.get() instanceof TraversalParent)
- this.attachHaltedTraverser = ((TraversalParent) this.collectingBarrierStep.get()).getLocalChildren().stream().filter(TraverserMapReduce::isBeyondElementId).findAny().isPresent();
+ this.collectingBarrierStep = ((CollectingBarrierStep<?>) traversalEndStep).clone();
+ this.comparator = this.collectingBarrierStep instanceof OrderGlobalStep ? new ChainedComparator<Comparable>(((OrderGlobalStep) this.collectingBarrierStep).getComparators()) : null;
+ if (this.collectingBarrierStep instanceof TraversalParent) {
+ this.attachHaltedTraverser = ((TraversalParent) this.collectingBarrierStep).getLocalChildren().stream().filter(TraversalHelper::isBeyondElementId).findAny().isPresent();
+ }
}
if (traversalEndStep instanceof RangeGlobalStep)
- this.rangeGlobalStep = Optional.of(((RangeGlobalStep) traversalEndStep).clone());
+ this.rangeGlobalStep = ((RangeGlobalStep) traversalEndStep).clone();
if (traversalEndStep instanceof TailGlobalStep)
- this.tailGlobalStep = Optional.of(((TailGlobalStep) traversalEndStep).clone());
+ this.tailGlobalStep = ((TailGlobalStep) traversalEndStep).clone();
if (traversalEndStep instanceof DedupGlobalStep)
this.dedupGlobal = true;
@@ -98,7 +114,7 @@ public final class TraverserMapReduce extends StaticMapReduce<Comparable, Traver
@Override
public boolean doStage(final Stage stage) {
- return stage.equals(Stage.MAP) || this.collectingBarrierStep.isPresent() || this.rangeGlobalStep.isPresent() || this.tailGlobalStep.isPresent() || this.dedupGlobal;
+ return stage.equals(Stage.MAP) || null != this.collectingBarrierStep || null != this.rangeGlobalStep || null != this.tailGlobalStep || this.dedupGlobal;
}
@Override
@@ -106,7 +122,7 @@ public final class TraverserMapReduce extends StaticMapReduce<Comparable, Traver
vertex.<TraverserSet<Object>>property(TraversalVertexProgram.HALTED_TRAVERSERS).ifPresent(traverserSet -> IteratorUtils.removeOnNext(traverserSet.iterator()).forEachRemaining(traverser -> {
if (this.attachHaltedTraverser && !(traverser.get() instanceof Edge))
traverser.attach(Attachable.Method.get(vertex));
- if (this.comparator.isPresent()) // TODO: I think we shouldn't ever single key it -- always double emit to load balance the servers.
+ if (null != this.comparator) // TODO: I think we shouldn't ever single key it -- always double emit to load balance the servers.
emitter.emit(traverser, traverser);
else
emitter.emit(traverser);
@@ -115,7 +131,7 @@ public final class TraverserMapReduce extends StaticMapReduce<Comparable, Traver
@Override
public Optional<Comparator<Comparable>> getMapKeySort() {
- return this.comparator;
+ return Optional.ofNullable(this.comparator);
}
@Override
@@ -134,20 +150,20 @@ public final class TraverserMapReduce extends StaticMapReduce<Comparable, Traver
@Override
public Iterator<Traverser<?>> generateFinalResult(final Iterator<KeyValue<Comparable, Traverser<?>>> keyValues) {
- if (this.collectingBarrierStep.isPresent()) {
+ if (null != this.collectingBarrierStep) {
final TraverserSet<?> traverserSet = new TraverserSet<>();
while (keyValues.hasNext()) {
traverserSet.add((Traverser.Admin) keyValues.next().getValue().asAdmin());
}
- this.collectingBarrierStep.get().barrierConsumer((TraverserSet) traverserSet);
+ this.collectingBarrierStep.barrierConsumer((TraverserSet) traverserSet);
return (Iterator) traverserSet.iterator();
- } else if (this.rangeGlobalStep.isPresent()) {
- final RangeGlobalStep<?> rangeGlobalStep = this.rangeGlobalStep.get();
+ } else if (null != this.rangeGlobalStep) {
+ final RangeGlobalStep<?> rangeGlobalStep = this.rangeGlobalStep;
rangeGlobalStep.setBypass(false);
rangeGlobalStep.addStarts(IteratorUtils.map(keyValues, keyValue -> (Traverser) keyValue.getValue()));
return (Iterator) rangeGlobalStep;
- } else if (this.tailGlobalStep.isPresent()) {
- final TailGlobalStep<?> tailGlobalStep = this.tailGlobalStep.get();
+ } else if (null != this.tailGlobalStep) {
+ final TailGlobalStep<?> tailGlobalStep = this.tailGlobalStep;
tailGlobalStep.setBypass(false);
tailGlobalStep.addStarts(IteratorUtils.map(keyValues, keyValue -> (Traverser) keyValue.getValue()));
return (Iterator) tailGlobalStep;
@@ -165,23 +181,4 @@ public final class TraverserMapReduce extends StaticMapReduce<Comparable, Traver
public String getMemoryKey() {
return TRAVERSERS;
}
-
- /////////////////////
- /////////////////////
-
- /*
- * THIS NEEDS TO GO INTO TRAVERSAL HELPER ONCE WE GET THIS ALL STRAIGHTENED OUT WITH THE INSTRUCTION SET OF GREMLIN (TODO:)
- */
- private static boolean isBeyondElementId(final Traversal.Admin<?, ?> traversal) {
- return (traversal instanceof TokenTraversal && !((TokenTraversal) traversal).getToken().equals(T.id)) || traversal instanceof ElementValueTraversal ||
- traversal.getSteps().stream()
- .filter(step -> step instanceof VertexStep ||
- step instanceof EdgeVertexStep ||
- step instanceof PropertiesStep ||
- step instanceof PropertyMapStep ||
- (step instanceof TraversalParent &&
- (((TraversalParent) step).getLocalChildren().stream().filter(TraverserMapReduce::isBeyondElementId).findAny().isPresent() ||
- ((TraversalParent) step).getGlobalChildren().stream().filter(TraverserMapReduce::isBeyondElementId).findAny().isPresent())))
- .findAny().isPresent();
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/TraversalHelper.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/TraversalHelper.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/TraversalHelper.java
index 1641cc8..24baef3 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/TraversalHelper.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/TraversalHelper.java
@@ -21,6 +21,8 @@ package org.apache.tinkerpop.gremlin.process.traversal.util;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.lambda.ElementValueTraversal;
+import org.apache.tinkerpop.gremlin.process.traversal.lambda.TokenTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.step.HasContainerHolder;
import org.apache.tinkerpop.gremlin.process.traversal.step.Scoping;
import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
@@ -31,8 +33,10 @@ import org.apache.tinkerpop.gremlin.process.traversal.step.filter.WherePredicate
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.WhereTraversalStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.EdgeVertexStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.GraphStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.LabelStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.MatchStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.PropertiesStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.PropertyMapStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.SelectOneStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.VertexStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.StartStep;
@@ -62,6 +66,24 @@ public final class TraversalHelper {
private TraversalHelper() {
}
+ public static boolean isBeyondElementId(final Traversal.Admin<?, ?> traversal) {
+ if (traversal instanceof TokenTraversal && !((TokenTraversal) traversal).getToken().equals(T.id))
+ return true;
+ if (traversal instanceof ElementValueTraversal)
+ return true;
+ else
+ return traversal.getSteps().stream()
+ .filter(step -> step instanceof VertexStep ||
+ step instanceof LabelStep ||
+ step instanceof EdgeVertexStep ||
+ step instanceof PropertiesStep ||
+ step instanceof PropertyMapStep ||
+ (step instanceof TraversalParent &&
+ (((TraversalParent) step).getLocalChildren().stream().filter(TraversalHelper::isBeyondElementId).findAny().isPresent() ||
+ ((TraversalParent) step).getGlobalChildren().stream().filter(TraversalHelper::isBeyondElementId).findAny().isPresent())))
+ .findAny().isPresent();
+ }
+
public static Class getLastElementClass(final Traversal.Admin<?, ?> traversal) {
Step<?, ?> currentStep = traversal.getEndStep();
while (!(currentStep instanceof EmptyStep)) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/util/TraversalHelperTest.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/util/TraversalHelperTest.java b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/util/TraversalHelperTest.java
index f0fc45a..3e6196f 100644
--- a/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/util/TraversalHelperTest.java
+++ b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/util/TraversalHelperTest.java
@@ -21,17 +21,14 @@ package org.apache.tinkerpop.gremlin.process.util;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Mutating;
-import org.apache.tinkerpop.gremlin.process.traversal.step.filter.DropStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.FilterStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.filter.LambdaFilterStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.filter.TraversalFilterStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.AddVertexStartStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.PropertiesStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.IdentityStep;
-import org.apache.tinkerpop.gremlin.process.traversal.util.DefaultTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
+import org.apache.tinkerpop.gremlin.process.traversal.util.DefaultTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.structure.PropertyType;
import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
@@ -39,7 +36,10 @@ import org.junit.Test;
import org.mockito.Mockito;
import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -48,6 +48,16 @@ import static org.junit.Assert.*;
public class TraversalHelperTest {
@Test
+ public void shouldIdentifyScopeOfAccess() {
+ assertFalse(TraversalHelper.isBeyondElementId(__.identity().asAdmin()));
+ assertFalse(TraversalHelper.isBeyondElementId(__.id().asAdmin()));
+ assertTrue(TraversalHelper.isBeyondElementId(__.label().asAdmin()));
+ assertTrue(TraversalHelper.isBeyondElementId(__.values("name").asAdmin()));
+ assertTrue(TraversalHelper.isBeyondElementId(__.outE("knows").asAdmin()));
+ // assertTrue(TraversalHelper.isBeyondElementId(((TraversalParent) __.order().asAdmin().getStartStep()).getLocalChildren().get(0)));
+ }
+
+ @Test
public void shouldNotFindStepOfClassInTraversal() {
final Traversal.Admin traversal = new DefaultTraversal<>(EmptyGraph.instance());
traversal.asAdmin().addStep(0, new HasStep(traversal));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/gremlin-groovy-test/src/main/java/org/apache/tinkerpop/gremlin/process/GroovyProcessComputerSuite.java
----------------------------------------------------------------------
diff --git a/gremlin-groovy-test/src/main/java/org/apache/tinkerpop/gremlin/process/GroovyProcessComputerSuite.java b/gremlin-groovy-test/src/main/java/org/apache/tinkerpop/gremlin/process/GroovyProcessComputerSuite.java
index d7c1ab3..f74e1dd 100644
--- a/gremlin-groovy-test/src/main/java/org/apache/tinkerpop/gremlin/process/GroovyProcessComputerSuite.java
+++ b/gremlin-groovy-test/src/main/java/org/apache/tinkerpop/gremlin/process/GroovyProcessComputerSuite.java
@@ -23,10 +23,6 @@ import org.apache.tinkerpop.gremlin.GraphManager;
import org.apache.tinkerpop.gremlin.groovy.loaders.SugarLoader;
import org.apache.tinkerpop.gremlin.groovy.util.SugarTestHelper;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
-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.clustering.peerpressure.PeerPressureVertexProgramTest;
-import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgramTest;
import org.apache.tinkerpop.gremlin.process.traversal.step.branch.GroovyBranchTest;
import org.apache.tinkerpop.gremlin.process.traversal.step.branch.GroovyChooseTest;
import org.apache.tinkerpop.gremlin.process.traversal.step.branch.GroovyLocalTest;
@@ -169,12 +165,6 @@ public class GroovyProcessComputerSuite extends ProcessComputerSuite {
GroovyStoreTest.Traversals.class,
GroovySubgraphTest.Traversals.class,
GroovyTreeTest.Traversals.class,
-
- // algorithms
- PageRankVertexProgramTest.class,
- PeerPressureVertexProgramTest.class,
- BulkLoaderVertexProgramTest.class,
- BulkDumperVertexProgramTest.class,
};
public GroovyProcessComputerSuite(final Class<?> klass, final RunnerBuilder builder) throws InitializationError {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
index f56b09b..23ef0a7 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
@@ -22,10 +22,15 @@ import org.apache.commons.configuration.Configuration;
import org.apache.tinkerpop.gremlin.ExceptionCoverage;
import org.apache.tinkerpop.gremlin.LoadGraphWith;
import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest;
+import org.apache.tinkerpop.gremlin.process.computer.clustering.peerpressure.PeerPressureVertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.step.sideEffect.mapreduce.TraverserMapReduce;
import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
import org.apache.tinkerpop.gremlin.process.computer.util.StaticVertexProgram;
import org.apache.tinkerpop.gremlin.process.traversal.P;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
@@ -43,6 +48,7 @@ import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;
@@ -1828,5 +1834,49 @@ public class GraphComputerTest extends AbstractGremlinProcessTest {
}
}
-
+ @Test
+ @LoadGraphWith(MODERN)
+ public void shouldSupportJobChaining() throws Exception {
+ final ComputerResult result1 = graphProvider.getGraphComputer(graph)
+ .program(PageRankVertexProgram.build().iterations(5).create(graph)).persist(GraphComputer.Persist.EDGES).result(GraphComputer.ResultGraph.NEW).submit().get();
+ final Graph graph1 = result1.graph();
+ final Memory memory1 = result1.memory();
+ assertEquals(5, memory1.getIteration());
+ assertEquals(6, graph1.traversal().V().count().next().intValue());
+ assertEquals(6, graph1.traversal().E().count().next().intValue());
+ assertEquals(6, graph1.traversal().V().values(PageRankVertexProgram.PAGE_RANK).count().next().intValue());
+ assertEquals(6, graph1.traversal().V().values(PageRankVertexProgram.EDGE_COUNT).count().next().intValue());
+ //
+ final ComputerResult result2 = graph1.compute(graphProvider.getGraphComputer(graph1).getClass())
+ .program(PeerPressureVertexProgram.build().maxIterations(4).create(graph1)).persist(GraphComputer.Persist.EDGES).result(GraphComputer.ResultGraph.NEW).submit().get();
+ final Graph graph2 = result2.graph();
+ final Memory memory2 = result2.memory();
+ assertTrue(memory2.getIteration() <= 4);
+ assertEquals(6, graph2.traversal().V().count().next().intValue());
+ assertEquals(6, graph2.traversal().E().count().next().intValue());
+ assertEquals(6, graph2.traversal().V().values(PeerPressureVertexProgram.CLUSTER).count().next().intValue());
+ assertEquals(6, graph2.traversal().V().values(PeerPressureVertexProgram.VOTE_STRENGTH).count().next().intValue());
+ assertEquals(6, graph2.traversal().V().values(PageRankVertexProgram.PAGE_RANK).count().next().intValue());
+ assertEquals(6, graph2.traversal().V().values(PageRankVertexProgram.EDGE_COUNT).count().next().intValue());
+ //
+ final ComputerResult result3 = graph2.compute(graphProvider.getGraphComputer(graph2).getClass())
+ .program(TraversalVertexProgram.build().traversal(graph2.traversal().V().groupCount("m").by(__.values(PageRankVertexProgram.PAGE_RANK).count()).label().asAdmin()).create(graph2)).persist(GraphComputer.Persist.EDGES).result(GraphComputer.ResultGraph.NEW).submit().get();
+ final Graph graph3 = result3.graph();
+ final Memory memory3 = result3.memory();
+ assertTrue(memory3.keys().contains("m"));
+ assertTrue(memory3.keys().contains(TraverserMapReduce.TRAVERSERS));
+ assertEquals(1, memory3.<Map<Long, Long>>get("m").size());
+ assertEquals(6, memory3.<Map<Long, Long>>get("m").get(1l).intValue());
+ List<Traverser<String>> traversers = IteratorUtils.list(memory3.<Iterator<Traverser<String>>>get(TraverserMapReduce.TRAVERSERS));
+ assertEquals(6, traversers.size());
+ assertEquals(4l, traversers.stream().map(Traverser::get).filter(s -> s.equals("person")).count());
+ assertEquals(2l, traversers.stream().map(Traverser::get).filter(s -> s.equals("software")).count());
+ assertEquals(6, graph3.traversal().V().count().next().intValue());
+ assertEquals(6, graph3.traversal().E().count().next().intValue());
+ assertEquals(6, graph3.traversal().V().values(TraversalVertexProgram.HALTED_TRAVERSERS).count().next().intValue());
+ assertEquals(6, graph3.traversal().V().values(PeerPressureVertexProgram.CLUSTER).count().next().intValue());
+ assertEquals(6, graph3.traversal().V().values(PeerPressureVertexProgram.VOTE_STRENGTH).count().next().intValue());
+ assertEquals(6, graph3.traversal().V().values(PageRankVertexProgram.PAGE_RANK).count().next().intValue());
+ assertEquals(6, graph3.traversal().V().values(PageRankVertexProgram.EDGE_COUNT).count().next().intValue());
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java
index e871269..364e71a 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java
@@ -58,24 +58,23 @@ public final class MapReduceHelper {
}
public static void executeMapReduceJob(final MapReduce mapReduce, final Memory.Admin memory, final Configuration configuration) throws IOException, ClassNotFoundException, InterruptedException {
- final boolean vertexProgramExists = configuration.get(VertexProgram.VERTEX_PROGRAM, null) != null;
final Configuration newConfiguration = new Configuration(configuration);
+ final boolean vertexProgramExists = newConfiguration.get(VertexProgram.VERTEX_PROGRAM, null) != null;
+ if (vertexProgramExists) {
+ newConfiguration.set(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputOutputHelper.getInputFormat((Class) newConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, OutputFormat.class)).getCanonicalName());
+ newConfiguration.unset(Constants.GREMLIN_HADOOP_GRAPH_FILTER);
+ }
final BaseConfiguration apacheConfiguration = new BaseConfiguration();
apacheConfiguration.setDelimiterParsingDisabled(true);
mapReduce.storeState(apacheConfiguration);
ConfUtil.mergeApacheIntoHadoopConfiguration(apacheConfiguration, newConfiguration);
if (!mapReduce.doStage(MapReduce.Stage.MAP)) {
- final Path memoryPath = new Path(Constants.getMemoryLocation(configuration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION), mapReduce.getMemoryKey()));
- mapReduce.addResultToMemory(memory, new ObjectWritableIterator(configuration, memoryPath));
+ final Path memoryPath = new Path(Constants.getMemoryLocation(newConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION), mapReduce.getMemoryKey()));
+ mapReduce.addResultToMemory(memory, new ObjectWritableIterator(newConfiguration, memoryPath));
} else {
final Optional<Comparator<?>> mapSort = mapReduce.getMapKeySort();
final Optional<Comparator<?>> reduceSort = mapReduce.getReduceKeySort();
-
newConfiguration.setClass(Constants.GREMLIN_HADOOP_MAP_REDUCE_CLASS, mapReduce.getClass(), MapReduce.class);
- if (vertexProgramExists) {
- newConfiguration.set(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputOutputHelper.getInputFormat((Class) newConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, OutputFormat.class)).getCanonicalName());
- newConfiguration.unset(Constants.GREMLIN_HADOOP_GRAPH_FILTER);
- }
final Job job = Job.getInstance(newConfiguration, mapReduce.toString());
HadoopGraph.LOGGER.info(Constants.GREMLIN_HADOOP_JOB_PREFIX + mapReduce.toString());
job.setJarByClass(HadoopGraph.class);
@@ -101,9 +100,13 @@ public final class MapReduceHelper {
job.setInputFormatClass(GraphFilterInputFormat.class);
job.setOutputFormatClass(SequenceFileOutputFormat.class);
// if there is no vertex program, then grab the graph from the input location
- final Path graphPath = vertexProgramExists ?
- new Path(Constants.getGraphLocation(newConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION))) :
- new Path(newConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION));
+ final Path graphPath;
+ if (vertexProgramExists) {
+ graphPath = new Path(Constants.getGraphLocation(newConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION)));
+ } else {
+ graphPath = new Path(newConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION));
+ }
+
Path memoryPath = new Path(Constants.getMemoryLocation(newConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION), (reduceSort.isPresent() ? mapReduce.getMemoryKey() + "-temp" : mapReduce.getMemoryKey())));
if (FileSystem.get(newConfiguration).exists(memoryPath)) {
FileSystem.get(newConfiguration).delete(memoryPath, true);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
index 4d8934f..4d9280f 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
@@ -139,7 +139,8 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
if (null == inputLocation)
inputLocation = hadoopConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION);
- if (hadoopConfiguration.get(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, null) == null && // if an InputRDD is specified, then ignore InputFormat
+ if (null != inputLocation &&
+ hadoopConfiguration.get(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, null) == null && // if an InputRDD is specified, then ignore InputFormat
hadoopConfiguration.get(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, null) != null &&
FileInputFormat.class.isAssignableFrom(hadoopConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputFormat.class))) {
try {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/OutputFormatRDD.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/OutputFormatRDD.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/OutputFormatRDD.java
index edb541b..b3fc91a 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/OutputFormatRDD.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/OutputFormatRDD.java
@@ -62,10 +62,11 @@ public final class OutputFormatRDD implements OutputRDD {
final String outputLocation = hadoopConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION);
if (null != outputLocation) {
// map back to a Hadoop stream for output
- memoryRDD.mapToPair(keyValue -> new Tuple2<>(new ObjectWritable<>(keyValue._1()), new ObjectWritable<>(keyValue._2()))).saveAsNewAPIHadoopFile(Constants.getMemoryLocation(outputLocation, memoryKey),
- ObjectWritable.class,
- ObjectWritable.class,
- SequenceFileOutputFormat.class, hadoopConfiguration);
+ memoryRDD.mapToPair(keyValue -> new Tuple2<>(new ObjectWritable<>(keyValue._1()), new ObjectWritable<>(keyValue._2())))
+ .saveAsNewAPIHadoopFile(Constants.getMemoryLocation(outputLocation, memoryKey),
+ ObjectWritable.class,
+ ObjectWritable.class,
+ SequenceFileOutputFormat.class, hadoopConfiguration);
try {
return (Iterator) new ObjectWritableIterator(hadoopConfiguration, new Path(Constants.getMemoryLocation(outputLocation, memoryKey)));
} catch (final IOException e) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/cd1a3794/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
index a6010b6..afa6906 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
@@ -28,6 +28,7 @@ import org.apache.tinkerpop.gremlin.hadoop.structure.io.FileSystemStorageCheck;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.engine.ComputerTraversalEngine;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.GroovyPageRankTest;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.PageRankTest;
import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedOutputRDD;
@@ -51,7 +52,14 @@ public final class SparkHadoopGraphProvider extends HadoopGraphProvider {
public Map<String, Object> getBaseConfiguration(final String graphName, final Class<?> test, final String testMethodName, final LoadGraphWith.GraphData loadGraphWith) {
final Map<String, Object> config = super.getBaseConfiguration(graphName, test, testMethodName, loadGraphWith);
config.put(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true); // this makes the test suite go really fast
- if (!test.equals(PageRankTest.Traversals.class) && !test.equals(FileSystemStorageCheck.class) && null != loadGraphWith && RANDOM.nextBoolean()) {
+
+ // toy graph inputRDD does not have corresponding outputRDD so where jobs chain, it fails (failing makes sense)
+ if (null != loadGraphWith &&
+ !test.equals(PageRankTest.Traversals.class) &&
+ !test.equals(GroovyPageRankTest.Traversals.class) &&
+ !test.equals(FileSystemStorageCheck.class) &&
+ !testMethodName.equals("shouldSupportJobChaining") && // GraphComputerTest.shouldSupportJobChaining
+ RANDOM.nextBoolean()) {
config.put(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, ToyGraphInputRDD.class.getCanonicalName());
}