TINKERPOP-1996 Fixed up general strategy application around io() The GraphComputer was not being set properly in the HadoopIoStep and therefore executions of OLAP runs would not work even if withComputer(SparkGraphComputer) was set. It only worked if the gremlin.hadoop.defaultGraphComputer property was set which was weird.
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/c97d747f Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/c97d747f Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/c97d747f Branch: refs/heads/TINKERPOP-1774 Commit: c97d747f4b05cef42eb0e53adae50fcdee083a2e Parents: 4d979cf Author: Stephen Mallette <sp...@genoprime.com> Authored: Fri Jul 20 10:30:00 2018 -0400 Committer: Stephen Mallette <sp...@genoprime.com> Committed: Fri Jul 20 10:30:00 2018 -0400 ---------------------------------------------------------------------- .../step/map/TraversalVertexProgramStep.java | 4 --- .../decoration/VertexProgramStrategy.java | 30 +++++++++----------- .../traversal/step/sideEffect/IoStep.java | 2 -- .../traversal/strategy/HadoopIoStrategy.java | 28 ++++++++++++------ 4 files changed, 33 insertions(+), 31 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/c97d747f/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java index 4eb950f..30cfee5 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java @@ -19,19 +19,15 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal.step.map; -import org.apache.tinkerpop.gremlin.process.computer.GraphComputer; import org.apache.tinkerpop.gremlin.process.computer.GraphFilter; import org.apache.tinkerpop.gremlin.process.computer.Memory; import org.apache.tinkerpop.gremlin.process.computer.traversal.MemoryTraversalSideEffects; import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram; -import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.VertexProgramStrategy; -import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.finalization.ComputerFinalizationStrategy; import org.apache.tinkerpop.gremlin.process.traversal.Traversal; import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies; import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy; import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent; import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement; -import org.apache.tinkerpop.gremlin.process.traversal.util.DefaultTraversalStrategies; import org.apache.tinkerpop.gremlin.process.traversal.util.PureTraversal; import org.apache.tinkerpop.gremlin.structure.Graph; import org.apache.tinkerpop.gremlin.structure.util.StringFactory; http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/c97d747f/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/VertexProgramStrategy.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/VertexProgramStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/VertexProgramStrategy.java index cb99652..fa6e23f 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/VertexProgramStrategy.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/VertexProgramStrategy.java @@ -101,23 +101,21 @@ public final class VertexProgramStrategy extends AbstractTraversalStrategy<Trave // wrap all non-VertexComputing steps into a TraversalVertexProgramStep currentStep = traversal.getStartStep(); - if (!(currentStep instanceof ReadWriting)) { + while (!(currentStep instanceof EmptyStep)) { + final Traversal.Admin<?, ?> computerTraversal = new DefaultTraversal<>(); + final Step<?, ?> firstLegalOLAPStep = getFirstLegalOLAPStep(currentStep); + final Step<?, ?> lastLegalOLAPStep = getLastLegalOLAPStep(currentStep); + if (!(firstLegalOLAPStep instanceof EmptyStep)) { + final int index = TraversalHelper.stepIndex(firstLegalOLAPStep, traversal); + TraversalHelper.removeToTraversal(firstLegalOLAPStep, lastLegalOLAPStep.getNextStep(), (Traversal.Admin) computerTraversal); + final TraversalVertexProgramStep traversalVertexProgramStep = new TraversalVertexProgramStep(traversal, computerTraversal); + traversal.addStep(index, traversalVertexProgramStep); + } + currentStep = traversal.getStartStep(); while (!(currentStep instanceof EmptyStep)) { - final Traversal.Admin<?, ?> computerTraversal = new DefaultTraversal<>(); - final Step<?, ?> firstLegalOLAPStep = getFirstLegalOLAPStep(currentStep); - final Step<?, ?> lastLegalOLAPStep = getLastLegalOLAPStep(currentStep); - if (!(firstLegalOLAPStep instanceof EmptyStep)) { - final int index = TraversalHelper.stepIndex(firstLegalOLAPStep, traversal); - TraversalHelper.removeToTraversal(firstLegalOLAPStep, lastLegalOLAPStep.getNextStep(), (Traversal.Admin) computerTraversal); - final TraversalVertexProgramStep traversalVertexProgramStep = new TraversalVertexProgramStep(traversal, computerTraversal); - traversal.addStep(index, traversalVertexProgramStep); - } - currentStep = traversal.getStartStep(); - while (!(currentStep instanceof EmptyStep)) { - if (!(currentStep instanceof VertexComputing)) - break; - currentStep = currentStep.getNextStep(); - } + if (!(currentStep instanceof VertexComputing)) + break; + currentStep = currentStep.getNextStep(); } } http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/c97d747f/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/IoStep.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/IoStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/IoStep.java index 1d4f40b..e5b3a6a 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/IoStep.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/IoStep.java @@ -18,8 +18,6 @@ */ package org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect; -import org.apache.commons.configuration.BaseConfiguration; -import org.apache.commons.configuration.Configuration; import org.apache.tinkerpop.gremlin.process.traversal.IO; import org.apache.tinkerpop.gremlin.process.traversal.Traversal; import org.apache.tinkerpop.gremlin.process.traversal.Traverser; http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/c97d747f/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/strategy/HadoopIoStrategy.java ---------------------------------------------------------------------- diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/strategy/HadoopIoStrategy.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/strategy/HadoopIoStrategy.java index 8348410..af3db6e 100644 --- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/strategy/HadoopIoStrategy.java +++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/strategy/HadoopIoStrategy.java @@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.hadoop.process.computer.traversal.strategy; import org.apache.tinkerpop.gremlin.hadoop.process.computer.traversal.step.sideEffect.HadoopIoStep; import org.apache.tinkerpop.gremlin.process.computer.clone.CloneVertexProgram; +import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep; import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.VertexProgramStep; import org.apache.tinkerpop.gremlin.process.traversal.Step; import org.apache.tinkerpop.gremlin.process.traversal.Traversal; @@ -30,6 +31,11 @@ import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.IoStep; import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy; import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + /** * The default implementation of the {@link IoStep} is a single threaded operation and doesn't properly take into * account the method by which OLAP read/writes take place with Hadoop. This strategy removes that step and replaces @@ -48,16 +54,20 @@ public final class HadoopIoStrategy extends AbstractTraversalStrategy<TraversalS @Override public void apply(final Traversal.Admin<?, ?> traversal) { - // replace IoStep steps with hadoop specific one - if (traversal.getStartStep() instanceof IoStep) { - final ReadWriting readWriting = (ReadWriting) traversal.getStartStep(); - final HadoopIoStep hadoopIoStep = new HadoopIoStep(traversal, readWriting.getFile()); - hadoopIoStep.setMode(readWriting.getMode()); - readWriting.getParameters().getRaw().entrySet().forEach(kv -> - hadoopIoStep.configure(kv.getKey(), kv.getValue().get(0)) - ); + // VertexProgramStrategy should wrap up the IoStep in a TraversalVertexProgramStep. use that to grab the + // GraphComputer that was injected in there and push that in to the HadoopIoStep. this step pattern match + // is fairly specific and since you really can't chain together steps after io() this approach should work + if (traversal.getStartStep() instanceof TraversalVertexProgramStep) { + final TraversalVertexProgramStep tvp = (TraversalVertexProgramStep) traversal.getStartStep(); + if (tvp.computerTraversal.get().getStartStep() instanceof ReadWriting) { + final ReadWriting readWriting = (ReadWriting) tvp.computerTraversal.get().getStartStep(); + final HadoopIoStep hadoopIoStep = new HadoopIoStep(traversal, readWriting.getFile()); + hadoopIoStep.setMode(readWriting.getMode()); + hadoopIoStep.setComputer(tvp.getComputer()); + readWriting.getParameters().getRaw().forEach((key, value) -> value.forEach(v -> hadoopIoStep.configure(key, v))); - TraversalHelper.replaceStep((Step) readWriting, hadoopIoStep, traversal); + TraversalHelper.replaceStep(tvp, hadoopIoStep, traversal); + } } }
