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/05/06 16:09:00 UTC
[06/18] incubator-tinkerpop git commit: lots of cleanup.
NativeInterceptor is now called VertexProgramInterceptor. Its generalized to
be used by any GraphComputer implementation. There exists
SparkVertexProgramInterceptor which extends VertexProgramInter
lots of cleanup. NativeInterceptor is now called VertexProgramInterceptor. Its generalized to be used by any GraphComputer implementation. There exists SparkVertexProgramInterceptor which extends VertexProgramInterceptor. Cleaned up naming conventions in the test cases. Found a bug in TraversalSource that was causing RemoteGraph computer tests to fail. SparkGraphComputer is cleaner with how interception happens -- really clear delineation which will be used for all GraphComptuers that use the Interceptor-model. Running integration tests over night.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/d5e7c786
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/d5e7c786
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/d5e7c786
Branch: refs/heads/master
Commit: d5e7c78628d89c464418f1329a797fc192880e8b
Parents: 7c103c8
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue May 3 20:55:03 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue May 3 20:55:03 2016 -0600
----------------------------------------------------------------------
.../traversal/strategy/TraversalMatcher.java | 47 ++++++++
.../strategy/VertexProgramInterceptor.java | 31 +++++
.../process/traversal/TraversalSource.java | 8 +-
.../tinkerpop/gremlin/hadoop/Constants.java | 2 +-
.../process/computer/NativeInterceptor.java | 32 ------
.../process/computer/SparkGraphComputer.java | 19 +--
.../optimization/SparkInterceptorStrategy.java | 60 ----------
.../SparkPartitionAwareStrategy.java | 86 --------------
.../interceptors/VertexCountInterceptor.java | 52 ---------
.../strategy/SparkVertexProgramInterceptor.java | 32 ++++++
.../optimization/SparkInterceptorStrategy.java | 60 ++++++++++
.../SparkPartitionAwareStrategy.java | 86 ++++++++++++++
.../SparkVertexCountInterceptor.java | 53 +++++++++
.../SparkInterceptorStrategyTest.java | 96 ----------------
.../SparkPartitionAwareStrategyTest.java | 115 -------------------
.../SparkInterceptorStrategyTest.java | 94 +++++++++++++++
.../SparkPartitionAwareStrategyTest.java | 113 ++++++++++++++++++
17 files changed, 536 insertions(+), 450 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/TraversalMatcher.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/TraversalMatcher.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/TraversalMatcher.java
new file mode 100644
index 0000000..cd37031
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/TraversalMatcher.java
@@ -0,0 +1,47 @@
+/*
+ * 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.traversal.strategy;
+
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.CountGlobalStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.GraphStep;
+
+import java.util.List;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class TraversalMatcher {
+
+ private TraversalMatcher() {
+ }
+
+ public static boolean is_g_V_count(final Traversal.Admin<?, ?> traversal) {
+ final List<Step> steps = traversal.getSteps();
+ return steps.size() == 2 &&
+ steps.get(0).getLabels().isEmpty() &&
+ steps.get(1).getLabels().isEmpty() &&
+ (steps.get(1) instanceof CountGlobalStep) &&
+ traversal.getSteps().get(0) instanceof GraphStep &&
+ ((GraphStep) traversal.getSteps().get(0)).returnsVertex() &&
+ ((GraphStep) traversal.getSteps().get(0)).getIds().length == 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/VertexProgramInterceptor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/VertexProgramInterceptor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/VertexProgramInterceptor.java
new file mode 100644
index 0000000..f174034
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/VertexProgramInterceptor.java
@@ -0,0 +1,31 @@
+/*
+ * 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.traversal.strategy;
+
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
+import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public interface VertexProgramInterceptor<V extends VertexProgram, G, M extends Memory> {
+
+ public G apply(final V vertexProgram, final G graph, final M memory);
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalSource.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalSource.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalSource.java
index 3b60940..cf0b51f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalSource.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalSource.java
@@ -98,7 +98,13 @@ public interface TraversalSource extends Cloneable {
* @return a new traversal source with updated strategies
*/
public default TraversalSource withComputer(final Computer computer) {
- final List<TraversalStrategy<?>> graphComputerStrategies = TraversalStrategies.GlobalCache.getStrategies(computer.apply(this.getGraph()).getClass()).toList();
+ Class<? extends GraphComputer> graphComputerClass;
+ try {
+ graphComputerClass = computer.apply(this.getGraph()).getClass();
+ } catch (final Exception e) {
+ graphComputerClass = computer.getGraphComputerClass();
+ }
+ final List<TraversalStrategy<?>> graphComputerStrategies = TraversalStrategies.GlobalCache.getStrategies(graphComputerClass).toList();
final TraversalStrategy[] traversalStrategies = new TraversalStrategy[graphComputerStrategies.size() + 1];
traversalStrategies[0] = new VertexProgramStrategy(computer);
for (int i = 0; i < graphComputerStrategies.size(); i++) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java
index ab45903..fb4f15a 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java
@@ -39,6 +39,7 @@ public final class Constants {
public static final String GREMLIN_HADOOP_GRAPH_WRITER_HAS_EDGES = "gremlin.hadoop.graphWriter.hasEdges";
public static final String GREMLIN_HADOOP_GRAPH_FILTER = "gremlin.hadoop.graphFilter";
public static final String GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER = "gremlin.hadoop.defaultGraphComputer";
+ public static final String GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR = "gremlin.hadoop.vertexProgramInterceptor";
public static final String GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE = "gremlin.hadoop.jarsInDistributedCache";
public static final String HIDDEN_G = Graph.Hidden.hide("g");
@@ -58,7 +59,6 @@ public final class Constants {
public static final String GREMLIN_SPARK_GRAPH_STORAGE_LEVEL = "gremlin.spark.graphStorageLevel";
public static final String GREMLIN_SPARK_PERSIST_STORAGE_LEVEL = "gremlin.spark.persistStorageLevel";
public static final String GREMLIN_SPARK_SKIP_PARTITIONER = "gremlin.spark.skipPartitioner";
- public static final String GREMLIN_SPARK_NATIVE_INTERCEPTOR = "gremlin.spark.nativeInterceptor";
public static final String SPARK_SERIALIZER = "spark.serializer";
public static String getGraphLocation(final String location) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/NativeInterceptor.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/NativeInterceptor.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/NativeInterceptor.java
deleted file mode 100644
index 8448a1b..0000000
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/NativeInterceptor.java
+++ /dev/null
@@ -1,32 +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.spark.process.computer;
-
-import org.apache.spark.api.java.JavaPairRDD;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
-import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public interface NativeInterceptor<V extends VertexProgram> {
-
- public JavaPairRDD<Object, VertexWritable> apply(final V vertexProgram, final JavaPairRDD<Object, VertexWritable> inputRDD, final SparkMemory memory);
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/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 49de96a..65e1c96 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
@@ -53,8 +53,9 @@ import org.apache.tinkerpop.gremlin.process.computer.util.DefaultComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies;
import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewIncomingPayload;
-import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.optimization.SparkInterceptorStrategy;
-import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.optimization.SparkPartitionAwareStrategy;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.SparkVertexProgramInterceptor;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.SparkInterceptorStrategy;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.SparkPartitionAwareStrategy;
import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.apache.tinkerpop.gremlin.spark.structure.io.InputFormatRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.InputOutputHelper;
@@ -241,16 +242,19 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
////////////////////////////////
if (null != this.vertexProgram) {
memory = new SparkMemory(this.vertexProgram, this.mapReducers, sparkContext);
- this.vertexProgram.setup(memory);
- if (apacheConfiguration.containsKey(Constants.GREMLIN_SPARK_NATIVE_INTERCEPTOR)) {
+ /////////////////
+ // if there is a registered VertexProgramInterceptor, use it to by pass the GraphComputer semantics
+ if (apacheConfiguration.containsKey(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR)) {
try {
- final NativeInterceptor<VertexProgram> interceptor = (NativeInterceptor) Class.forName(apacheConfiguration.getString(Constants.GREMLIN_SPARK_NATIVE_INTERCEPTOR)).newInstance();
+ final SparkVertexProgramInterceptor<VertexProgram> interceptor =
+ (SparkVertexProgramInterceptor) Class.forName(apacheConfiguration.getString(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR)).newInstance();
computedGraphRDD = interceptor.apply(this.vertexProgram, loadedGraphRDD, memory);
} catch (final ClassNotFoundException | IllegalAccessException | InstantiationException e) {
throw new IllegalStateException(e.getMessage());
}
- } else {
+ } else { // standard GraphComputer semantics
// set up the vertex program and wire up configurations
+ this.vertexProgram.setup(memory);
JavaPairRDD<Object, ViewIncomingPayload<Object>> viewIncomingRDD = null;
memory.broadcastMemory(sparkContext);
final HadoopConfiguration vertexProgramConfiguration = new HadoopConfiguration();
@@ -272,6 +276,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
// write the computed graph to the respective output (rdd or output format)
computedGraphRDD = SparkExecutor.prepareFinalGraphRDD(loadedGraphRDD, viewIncomingRDD, this.vertexProgram.getVertexComputeKeys());
}
+ /////////////////
memory.complete(); // drop all transient memory keys
if (null != outputRDD && !this.persist.equals(Persist.NOTHING)) {
outputRDD.writeGraphRDD(apacheConfiguration, computedGraphRDD);
@@ -318,7 +323,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
// unpersist the loaded graph if it will not be used again (no PersistedInputRDD)
// if the graphRDD was loaded from Spark, but then partitioned, its a different RDD
- if ((!inputFromSpark || partitioned || filtered) && computedGraphCreated)
+ if ((!skipPartitioner && (!inputFromSpark || partitioned || filtered)) && computedGraphCreated)
loadedGraphRDD.unpersist();
// unpersist the computed graph if it will not be used again (no PersistedOutputRDD)
if (!outputToSpark || this.persist.equals(GraphComputer.Persist.NOTHING))
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkInterceptorStrategy.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkInterceptorStrategy.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkInterceptorStrategy.java
deleted file mode 100644
index 2ad1ad6..0000000
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkInterceptorStrategy.java
+++ /dev/null
@@ -1,60 +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.spark.process.computer.traversal.optimization;
-
-import org.apache.tinkerpop.gremlin.hadoop.Constants;
-import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
-import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.optimization.interceptors.VertexCountInterceptor;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
-
-import java.util.List;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class SparkInterceptorStrategy extends AbstractTraversalStrategy<TraversalStrategy.ProviderOptimizationStrategy> implements TraversalStrategy.ProviderOptimizationStrategy {
-
- private static final SparkInterceptorStrategy INSTANCE = new SparkInterceptorStrategy();
-
- private SparkInterceptorStrategy() {
- }
-
- @Override
- public void apply(final Traversal.Admin<?, ?> traversal) {
- final Graph graph = traversal.getGraph().orElse(EmptyGraph.instance());
- final List<TraversalVertexProgramStep> steps = TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal);
- for (final TraversalVertexProgramStep step : steps) {
- if (step.generateProgram(graph).getTraversal().equals(graph.traversal().V().count())) {
- step.setComputer(step.getComputer().configure(Constants.GREMLIN_SPARK_NATIVE_INTERCEPTOR, VertexCountInterceptor.class.getCanonicalName()));
- }
- }
- }
-
- public static SparkInterceptorStrategy instance() {
- return INSTANCE;
- }
-
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkPartitionAwareStrategy.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkPartitionAwareStrategy.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkPartitionAwareStrategy.java
deleted file mode 100644
index 5c28d41..0000000
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkPartitionAwareStrategy.java
+++ /dev/null
@@ -1,86 +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.spark.process.computer.traversal.optimization;
-
-import org.apache.tinkerpop.gremlin.hadoop.Constants;
-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.TraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.EdgeVertexStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.LambdaFlatMapStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.LambdaMapStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.VertexStep;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
-import org.apache.tinkerpop.gremlin.structure.Direction;
-import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
-
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class SparkPartitionAwareStrategy extends AbstractTraversalStrategy<TraversalStrategy.ProviderOptimizationStrategy> implements TraversalStrategy.ProviderOptimizationStrategy {
-
- private static final SparkPartitionAwareStrategy INSTANCE = new SparkPartitionAwareStrategy();
-
- private static final Set<Class<? extends Step>> MESSAGE_PASS_CLASSES = new HashSet<>(Arrays.asList(
- EdgeVertexStep.class,
- LambdaMapStep.class, // maybe?
- LambdaFlatMapStep.class // maybe?
- // VertexStep is special as you need to see if the return class is Edge or Vertex (logic below)
- ));
-
- private SparkPartitionAwareStrategy() {
- }
-
- @Override
- public void apply(final Traversal.Admin<?, ?> traversal) {
- final List<TraversalVertexProgramStep> steps = TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal);
- for (final TraversalVertexProgramStep step : steps) {
- final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(traversal.getGraph().orElse(EmptyGraph.instance())).getTraversal();
- boolean messagePasses = MESSAGE_PASS_CLASSES.stream()
- .flatMap(clazz -> TraversalHelper.<Step<?, ?>>getStepsOfAssignableClassRecursively((Class) clazz, computerTraversal).stream())
- .filter(s -> TraversalHelper.isGlobalChild(((Step) s).getTraversal().asAdmin()))
- .findAny()
- .isPresent();
- if (!messagePasses) {
- for (final VertexStep vertexStep : TraversalHelper.getStepsOfAssignableClassRecursively(VertexStep.class, computerTraversal)) {
- if (TraversalHelper.isGlobalChild(vertexStep.getTraversal()) &&
- (vertexStep.returnsVertex() || !vertexStep.getDirection().equals(Direction.OUT))) { // in edges require message pass in OLAP
- messagePasses = true;
- break;
- }
- }
- }
- if (!messagePasses) // if no message passing, don't partition (save time)
- step.setComputer(step.getComputer().configure(Constants.GREMLIN_SPARK_SKIP_PARTITIONER, true));
- }
- }
-
- public static SparkPartitionAwareStrategy instance() {
- return INSTANCE;
- }
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/interceptors/VertexCountInterceptor.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/interceptors/VertexCountInterceptor.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/interceptors/VertexCountInterceptor.java
deleted file mode 100644
index b685677..0000000
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/interceptors/VertexCountInterceptor.java
+++ /dev/null
@@ -1,52 +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.spark.process.computer.traversal.optimization.interceptors;
-
-import org.apache.spark.api.java.JavaPairRDD;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
-import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
-import org.apache.tinkerpop.gremlin.process.traversal.Step;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
-import org.apache.tinkerpop.gremlin.spark.process.computer.NativeInterceptor;
-import org.apache.tinkerpop.gremlin.spark.process.computer.SparkMemory;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class VertexCountInterceptor implements NativeInterceptor<TraversalVertexProgram> {
-
- public VertexCountInterceptor() {
-
- }
-
- @Override
- public JavaPairRDD<Object, VertexWritable> apply(final TraversalVertexProgram vertexProgram, final JavaPairRDD<Object, VertexWritable> inputRDD, final SparkMemory memory) {
- final Traversal.Admin<Vertex, Long> traversal = (Traversal.Admin) vertexProgram.getTraversal();
- final TraverserSet<Long> haltedTraversers = new TraverserSet<>();
- haltedTraversers.add(traversal.getTraverserGenerator().generate(inputRDD.count(), (Step) traversal.getEndStep(), 1l));
- memory.set(TraversalVertexProgram.HALTED_TRAVERSERS, haltedTraversers);
- memory.incrIteration();
- return inputRDD;
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/SparkVertexProgramInterceptor.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/SparkVertexProgramInterceptor.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/SparkVertexProgramInterceptor.java
new file mode 100644
index 0000000..0e34e45
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/SparkVertexProgramInterceptor.java
@@ -0,0 +1,32 @@
+/*
+ * 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.spark.process.computer.traversal.strategy;
+
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.VertexProgramInterceptor;
+import org.apache.tinkerpop.gremlin.spark.process.computer.SparkMemory;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public interface SparkVertexProgramInterceptor<V extends VertexProgram> extends VertexProgramInterceptor<V, JavaPairRDD<Object, VertexWritable>, SparkMemory> {
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategy.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategy.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategy.java
new file mode 100644
index 0000000..a22cef7
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategy.java
@@ -0,0 +1,60 @@
+/*
+ * 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.spark.process.computer.traversal.strategy.optimization;
+
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.TraversalMatcher;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.interceptor.SparkVertexCountInterceptor;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
+
+import java.util.List;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class SparkInterceptorStrategy extends AbstractTraversalStrategy<TraversalStrategy.ProviderOptimizationStrategy> implements TraversalStrategy.ProviderOptimizationStrategy {
+
+ private static final SparkInterceptorStrategy INSTANCE = new SparkInterceptorStrategy();
+
+ private SparkInterceptorStrategy() {
+ }
+
+ @Override
+ public void apply(final Traversal.Admin<?, ?> traversal) {
+ final Graph graph = traversal.getGraph().orElse(EmptyGraph.instance());
+ final List<TraversalVertexProgramStep> steps = TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal);
+ for (final TraversalVertexProgramStep step : steps) {
+ if (TraversalMatcher.is_g_V_count(step.generateProgram(graph).getTraversal()))
+ step.setComputer(step.getComputer().configure(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR, SparkVertexCountInterceptor.class.getCanonicalName()));
+ }
+ }
+
+ public static SparkInterceptorStrategy instance() {
+ return INSTANCE;
+ }
+
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkPartitionAwareStrategy.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkPartitionAwareStrategy.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkPartitionAwareStrategy.java
new file mode 100644
index 0000000..de3e737
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkPartitionAwareStrategy.java
@@ -0,0 +1,86 @@
+/*
+ * 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.spark.process.computer.traversal.strategy.optimization;
+
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+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.TraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.EdgeVertexStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.LambdaFlatMapStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.LambdaMapStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.VertexStep;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
+import org.apache.tinkerpop.gremlin.structure.Direction;
+import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class SparkPartitionAwareStrategy extends AbstractTraversalStrategy<TraversalStrategy.ProviderOptimizationStrategy> implements TraversalStrategy.ProviderOptimizationStrategy {
+
+ private static final SparkPartitionAwareStrategy INSTANCE = new SparkPartitionAwareStrategy();
+
+ private static final Set<Class<? extends Step>> MESSAGEPASS_CLASSES = new HashSet<>(Arrays.asList(
+ EdgeVertexStep.class,
+ LambdaMapStep.class, // maybe?
+ LambdaFlatMapStep.class // maybe?
+ // VertexStep is special as you need to see if the return class is Edge or Vertex (logic below)
+ ));
+
+ private SparkPartitionAwareStrategy() {
+ }
+
+ @Override
+ public void apply(final Traversal.Admin<?, ?> traversal) {
+ final List<TraversalVertexProgramStep> steps = TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal);
+ for (final TraversalVertexProgramStep step : steps) {
+ final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(traversal.getGraph().orElse(EmptyGraph.instance())).getTraversal();
+ boolean messagePasses = MESSAGEPASS_CLASSES.stream()
+ .flatMap(clazz -> TraversalHelper.<Step<?, ?>>getStepsOfAssignableClassRecursively((Class) clazz, computerTraversal).stream())
+ .filter(s -> TraversalHelper.isGlobalChild(((Step) s).getTraversal().asAdmin()))
+ .findAny()
+ .isPresent();
+ if (!messagePasses) {
+ for (final VertexStep vertexStep : TraversalHelper.getStepsOfAssignableClassRecursively(VertexStep.class, computerTraversal)) {
+ if (TraversalHelper.isGlobalChild(vertexStep.getTraversal()) &&
+ (vertexStep.returnsVertex() || !vertexStep.getDirection().equals(Direction.OUT))) { // in edges require message pass in OLAP
+ messagePasses = true;
+ break;
+ }
+ }
+ }
+ if (!messagePasses) // if no message passing, don't partition (save time)
+ step.setComputer(step.getComputer().configure(Constants.GREMLIN_SPARK_SKIP_PARTITIONER, true));
+ }
+ }
+
+ public static SparkPartitionAwareStrategy instance() {
+ return INSTANCE;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkVertexCountInterceptor.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkVertexCountInterceptor.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkVertexCountInterceptor.java
new file mode 100644
index 0000000..8feb628
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkVertexCountInterceptor.java
@@ -0,0 +1,53 @@
+/*
+ * 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.spark.process.computer.traversal.strategy.optimization.interceptor;
+
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
+import org.apache.tinkerpop.gremlin.spark.process.computer.SparkMemory;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.SparkVertexProgramInterceptor;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class SparkVertexCountInterceptor implements SparkVertexProgramInterceptor<TraversalVertexProgram> {
+
+ public SparkVertexCountInterceptor() {
+
+ }
+
+ @Override
+ public JavaPairRDD<Object, VertexWritable> apply(final TraversalVertexProgram vertexProgram, final JavaPairRDD<Object, VertexWritable> inputRDD, final SparkMemory memory) {
+ vertexProgram.setup(memory);
+ final Traversal.Admin<Vertex, Long> traversal = (Traversal.Admin) vertexProgram.getTraversal();
+ final TraverserSet<Long> haltedTraversers = new TraverserSet<>();
+ haltedTraversers.add(traversal.getTraverserGenerator().generate(inputRDD.count(), (Step) traversal.getEndStep(), 1l));
+ memory.set(TraversalVertexProgram.HALTED_TRAVERSERS, haltedTraversers);
+ memory.incrIteration();
+ return inputRDD;
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkInterceptorStrategyTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkInterceptorStrategyTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkInterceptorStrategyTest.java
deleted file mode 100644
index bb009eb..0000000
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkInterceptorStrategyTest.java
+++ /dev/null
@@ -1,96 +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.spark.process.computer.traversal.optimization;
-
-import org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.TestHelper;
-import org.apache.tinkerpop.gremlin.hadoop.Constants;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat;
-import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
-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.util.TraversalHelper;
-import org.apache.tinkerpop.gremlin.spark.AbstractSparkTest;
-import org.apache.tinkerpop.gremlin.spark.process.computer.SparkGraphComputer;
-import org.apache.tinkerpop.gremlin.spark.process.computer.SparkHadoopGraphProvider;
-import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.optimization.interceptors.VertexCountInterceptor;
-import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedOutputRDD;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
-import org.junit.Test;
-
-import java.util.UUID;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class SparkInterceptorStrategyTest extends AbstractSparkTest {
-
- @Test
- public void shouldUseVertexCountInterceptor() throws Exception {
- final String outputLocation = TestHelper.makeTestDataDirectory(SparkPartitionAwareStrategyTest.class, UUID.randomUUID().toString());
- Configuration configuration = getBaseConfiguration();
- configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
- configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, outputLocation);
- configuration.setProperty(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
- ///
- Graph graph = GraphFactory.open(configuration);
- GraphTraversalSource g = graph.traversal().withComputer();
- assertTrue(g.getStrategies().toList().contains(SparkInterceptorStrategy.instance()));
- assertTrue(g.V().count().explain().toString().contains(SparkInterceptorStrategy.class.getSimpleName()));
- //
- assertEquals(6l, g.V().count().next().longValue());
- assertEquals(2l, g.V().out().out().count().next().longValue());
- }
-
- @Test
- public void shouldSetConfigurationsCorrectly() throws Exception {
- final String outputLocation = TestHelper.makeTestDataDirectory(SparkPartitionAwareStrategyTest.class, UUID.randomUUID().toString());
- Configuration configuration = getBaseConfiguration();
- configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
- configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, outputLocation);
- configuration.setProperty(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
-
- Graph graph = GraphFactory.open(configuration);
- GraphTraversalSource g = graph.traversal().withComputer();
-
- assertEquals(VertexCountInterceptor.class.getCanonicalName(), getInterceptor(g.V().count()));
- assertEquals(VertexCountInterceptor.class.getCanonicalName(), getInterceptor(g.V().identity().identity().count()));
- assertNull(getInterceptor(g.V().out().count()));
- assertNull(getInterceptor(g.V().as("a").count()));
- }
-
- private static String getInterceptor(final Traversal<?, ?> traversal) {
- traversal.asAdmin().applyStrategies();
- return (String) TraversalHelper.getFirstStepOfAssignableClass(TraversalVertexProgramStep.class, traversal.asAdmin()).get()
- .getComputer()
- .getConfiguration()
- .getOrDefault(Constants.GREMLIN_SPARK_NATIVE_INTERCEPTOR, null);
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkPartitionAwareStrategyTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkPartitionAwareStrategyTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkPartitionAwareStrategyTest.java
deleted file mode 100644
index 4324202..0000000
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/optimization/SparkPartitionAwareStrategyTest.java
+++ /dev/null
@@ -1,115 +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.spark.process.computer.traversal.optimization;
-
-import org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.TestHelper;
-import org.apache.tinkerpop.gremlin.hadoop.Constants;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat;
-import org.apache.tinkerpop.gremlin.process.computer.Computer;
-import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
-import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
-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.process.traversal.util.TraversalHelper;
-import org.apache.tinkerpop.gremlin.spark.AbstractSparkTest;
-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.io.PersistedOutputRDD;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
-import org.junit.Test;
-
-import java.util.UUID;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class SparkPartitionAwareStrategyTest extends AbstractSparkTest {
-
- @Test
- public void shouldPartitionTheInputRDDAccordingly() throws Exception {
- final String outputLocation = TestHelper.makeTestDataDirectory(SparkPartitionAwareStrategyTest.class, UUID.randomUUID().toString());
- Configuration configuration = getBaseConfiguration();
- configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
- configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, outputLocation);
- configuration.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
- configuration.setProperty(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
- ///
- Graph graph = GraphFactory.open(configuration);
- GraphTraversalSource g = graph.traversal().withComputer();
- assertTrue(g.getStrategies().toList().contains(SparkPartitionAwareStrategy.instance()));
- assertTrue(g.V().count().explain().toString().contains(SparkPartitionAwareStrategy.class.getSimpleName()));
- //
- assertEquals(6l, g.V().count().next().longValue());
- assertEquals(2l, g.V().out().out().count().next().longValue());
- }
-
-
- @Test
- public void shouldSetConfigurationsCorrectly() throws Exception {
- final String outputLocation = TestHelper.makeTestDataDirectory(SparkPartitionAwareStrategyTest.class, UUID.randomUUID().toString());
- Configuration configuration = getBaseConfiguration();
- configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
- configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, outputLocation);
- configuration.setProperty(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
-
- Graph graph = GraphFactory.open(configuration);
- GraphTraversalSource g = graph.traversal().withComputer(Computer.compute().persist(GraphComputer.Persist.EDGES).result(GraphComputer.ResultGraph.NEW));
-
- assertTrue(skipPartitioner(g.V().limit(10)));
- assertTrue(skipPartitioner(g.V().values("age").groupCount()));
- assertTrue(skipPartitioner(g.V().groupCount().by(__.out().count())));
- assertTrue(skipPartitioner(g.V().outE()));
- assertTrue(skipPartitioner(g.V().count()));
- assertTrue(skipPartitioner(g.V().out().count()));
- assertTrue(skipPartitioner(g.V().local(__.inE()).count()));
- assertTrue(skipPartitioner(g.V().outE().inV().count()));
- ////
- assertFalse(skipPartitioner(g.V().outE().inV()));
- assertFalse(skipPartitioner(g.V().both()));
- assertFalse(skipPartitioner(g.V().both().count()));
- assertFalse(skipPartitioner(g.V().out().id()));
- assertFalse(skipPartitioner(g.V().out().out().count()));
- assertFalse(skipPartitioner(g.V().in().count()));
- assertFalse(skipPartitioner(g.V().inE().count()));
-
-
- }
-
- private static boolean skipPartitioner(final Traversal<?, ?> traversal) {
- traversal.asAdmin().applyStrategies();
- return (Boolean) TraversalHelper.getFirstStepOfAssignableClass(TraversalVertexProgramStep.class, traversal.asAdmin()).get()
- .getComputer()
- .getConfiguration()
- .getOrDefault(Constants.GREMLIN_SPARK_SKIP_PARTITIONER, false);
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategyTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategyTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategyTest.java
new file mode 100644
index 0000000..1e73372
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkInterceptorStrategyTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.spark.process.computer.traversal.strategy.optimization;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.TestHelper;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
+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.util.TraversalHelper;
+import org.apache.tinkerpop.gremlin.spark.AbstractSparkTest;
+import org.apache.tinkerpop.gremlin.spark.process.computer.SparkGraphComputer;
+import org.apache.tinkerpop.gremlin.spark.process.computer.SparkHadoopGraphProvider;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.interceptor.SparkVertexCountInterceptor;
+import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedOutputRDD;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
+import org.junit.Test;
+
+import java.util.UUID;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class SparkInterceptorStrategyTest extends AbstractSparkTest {
+
+ @Test
+ public void shouldSuccessfullyEvaluateInterceptedTraversals() throws Exception {
+ final Configuration configuration = getBaseConfiguration();
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, TestHelper.makeTestDataDirectory(SparkPartitionAwareStrategyTest.class, UUID.randomUUID().toString()));
+ configuration.setProperty(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
+ ///
+ Graph graph = GraphFactory.open(configuration);
+ GraphTraversalSource g = graph.traversal().withComputer();
+ assertTrue(g.getStrategies().toList().contains(SparkInterceptorStrategy.instance()));
+ assertTrue(g.V().count().explain().toString().contains(SparkInterceptorStrategy.class.getSimpleName()));
+ //
+ assertEquals(6l, g.V().count().next().longValue());
+ assertEquals(2l, g.V().out().out().count().next().longValue());
+ }
+
+ @Test
+ public void shouldInterceptExceptedTraversals() throws Exception {
+ final Configuration configuration = getBaseConfiguration();
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, TestHelper.makeTestDataDirectory(SparkPartitionAwareStrategyTest.class, UUID.randomUUID().toString()));
+ configuration.setProperty(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
+ ///
+ Graph graph = GraphFactory.open(configuration);
+ GraphTraversalSource g = graph.traversal().withComputer();
+ //
+ assertEquals(SparkVertexCountInterceptor.class.getCanonicalName(), getInterceptor(g.V().count()));
+ assertEquals(SparkVertexCountInterceptor.class.getCanonicalName(), getInterceptor(g.V().identity().identity().count()));
+ assertNull(getInterceptor(g.V().out().count()));
+ assertNull(getInterceptor(g.V().as("a").count()));
+ }
+
+ private static String getInterceptor(final Traversal<?, ?> traversal) {
+ traversal.asAdmin().applyStrategies();
+ return (String) TraversalHelper.getFirstStepOfAssignableClass(TraversalVertexProgramStep.class, traversal.asAdmin()).get()
+ .getComputer()
+ .getConfiguration()
+ .getOrDefault(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR, null);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d5e7c786/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkPartitionAwareStrategyTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkPartitionAwareStrategyTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkPartitionAwareStrategyTest.java
new file mode 100644
index 0000000..045f424
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkPartitionAwareStrategyTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.spark.process.computer.traversal.strategy.optimization;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.TestHelper;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
+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.process.traversal.util.TraversalHelper;
+import org.apache.tinkerpop.gremlin.spark.AbstractSparkTest;
+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.io.PersistedOutputRDD;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
+import org.junit.Test;
+
+import java.util.UUID;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class SparkPartitionAwareStrategyTest extends AbstractSparkTest {
+
+ @Test
+ public void shouldSuccessfullyEvaluateSkipPartitionedTraversals() throws Exception {
+ final String outputLocation = TestHelper.makeTestDataDirectory(SparkPartitionAwareStrategyTest.class, UUID.randomUUID().toString());
+ Configuration configuration = getBaseConfiguration();
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, outputLocation);
+ configuration.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
+ configuration.setProperty(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
+ ///
+ Graph graph = GraphFactory.open(configuration);
+ GraphTraversalSource g = graph.traversal().withComputer();
+ assertTrue(g.getStrategies().toList().contains(SparkPartitionAwareStrategy.instance()));
+ assertTrue(g.V().count().explain().toString().contains(SparkPartitionAwareStrategy.class.getSimpleName()));
+ //
+ assertEquals(6l, g.V().count().next().longValue());
+ assertEquals(2l, g.V().out().out().count().next().longValue());
+ }
+
+
+ @Test
+ public void shouldSkipPartitionExceptedTraversals() throws Exception {
+ final String outputLocation = TestHelper.makeTestDataDirectory(SparkPartitionAwareStrategyTest.class, UUID.randomUUID().toString());
+ Configuration configuration = getBaseConfiguration();
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, outputLocation);
+ configuration.setProperty(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
+
+ Graph graph = GraphFactory.open(configuration);
+ GraphTraversalSource g = graph.traversal().withComputer();
+
+ assertTrue(skipPartitioner(g.V().limit(10)));
+ assertTrue(skipPartitioner(g.V().values("age").groupCount()));
+ assertTrue(skipPartitioner(g.V().groupCount().by(__.out().count())));
+ assertTrue(skipPartitioner(g.V().outE()));
+ assertTrue(skipPartitioner(g.V().count()));
+ assertTrue(skipPartitioner(g.V().out().count()));
+ assertTrue(skipPartitioner(g.V().local(__.inE()).count()));
+ assertTrue(skipPartitioner(g.V().outE().inV().count()));
+ ////
+ assertFalse(skipPartitioner(g.V().outE().inV()));
+ assertFalse(skipPartitioner(g.V().both()));
+ assertFalse(skipPartitioner(g.V().both().count()));
+ assertFalse(skipPartitioner(g.V().out().id()));
+ assertFalse(skipPartitioner(g.V().out().out().count()));
+ assertFalse(skipPartitioner(g.V().in().count()));
+ assertFalse(skipPartitioner(g.V().inE().count()));
+
+
+ }
+
+ private static boolean skipPartitioner(final Traversal<?, ?> traversal) {
+ traversal.asAdmin().applyStrategies();
+ return (Boolean) TraversalHelper.getFirstStepOfAssignableClass(TraversalVertexProgramStep.class, traversal.asAdmin()).get()
+ .getComputer()
+ .getConfiguration()
+ .getOrDefault(Constants.GREMLIN_SPARK_SKIP_PARTITIONER, false);
+
+ }
+
+}