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 2015/03/18 18:43:23 UTC
incubator-tinkerpop git commit: Removed ComputerDataStrategy and
replaced it with ComputerGraph. This, as a crazy sideEffect,
solved a nasty serialization issue I was having with SparkGraphComputer.
Finally, added ChainedComparator to ensure chained comp
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master 98c65988d -> c6fef8f51
Removed ComputerDataStrategy and replaced it with ComputerGraph. This, as a crazy sideEffect, solved a nasty serialization issue I was having with SparkGraphComputer. Finally, added ChainedComparator to ensure chained comparators are serializable.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/c6fef8f5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/c6fef8f5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/c6fef8f5
Branch: refs/heads/master
Commit: c6fef8f512c618f0f577601b1cfa84e31c16745e
Parents: 98c6598
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed Mar 18 11:43:07 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed Mar 18 11:43:20 2015 -0600
----------------------------------------------------------------------
.../mapreduce/TraverserMapReduce.java | 5 +-
.../computer/util/ComputerDataStrategy.java | 79 -----
.../process/computer/util/ComputerGraph.java | 344 +++++++++++++++++++
.../computer/util/GraphComputerHelper.java | 16 +-
.../traversal/step/map/OrderGlobalStep.java | 13 +-
.../gremlin/process/util/TraverserSet.java | 2 +-
.../util/function/ChainedComparator.java | 44 +++
.../gremlin/process/ProcessComputerSuite.java | 5 +-
.../computer/util/ComputerDataStrategyTest.java | 88 -----
.../computer/util/ComputerGraphTest.java | 89 +++++
hadoop-gremlin/conf/hadoop-gryo.properties | 2 +
.../computer/giraph/GiraphComputeVertex.java | 13 +-
.../computer/giraph/GiraphGraphComputer.java | 5 +-
.../computer/spark/SparkGraphComputer.java | 5 +-
.../computer/spark/util/SparkHelper.java | 5 +-
.../process/computer/TinkerGraphComputer.java | 15 +-
16 files changed, 516 insertions(+), 214 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/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 0048531..041f34d 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
@@ -30,6 +30,7 @@ import org.apache.tinkerpop.gremlin.process.util.TraverserSet;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
+import org.apache.tinkerpop.gremlin.util.function.ChainedComparator;
import java.util.Comparator;
import java.util.Iterator;
@@ -50,14 +51,14 @@ public final class TraverserMapReduce extends StaticMapReduce<Comparable, Object
public TraverserMapReduce(final Step traversalEndStep) {
this.traversal = traversalEndStep.getTraversal();
- this.comparator = Optional.ofNullable(traversalEndStep instanceof ComparatorHolder ? GraphComputerHelper.chainComparators(((ComparatorHolder) traversalEndStep).getComparators()) : null);
+ this.comparator = Optional.ofNullable(traversalEndStep instanceof ComparatorHolder ? new ChainedComparator<Comparable>(((ComparatorHolder) traversalEndStep).getComparators()) : null);
}
@Override
public void loadState(final Configuration configuration) {
this.traversal = TraversalVertexProgram.getTraversalSupplier(configuration).get();
final Step endStep = this.traversal.getEndStep().getPreviousStep(); // don't get the ComputerResultStep
- this.comparator = Optional.ofNullable(endStep instanceof ComparatorHolder ? GraphComputerHelper.chainComparators(((ComparatorHolder) endStep).getComparators()) : null);
+ this.comparator = Optional.ofNullable(endStep instanceof ComparatorHolder ? new ChainedComparator<Comparable>(((ComparatorHolder) endStep).getComparators()) : null);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerDataStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerDataStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerDataStrategy.java
deleted file mode 100644
index 08010b8..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerDataStrategy.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.tinkerpop.gremlin.process.computer.util;
-
-import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.structure.strategy.GraphStrategy;
-import org.apache.tinkerpop.gremlin.structure.strategy.StrategyContext;
-import org.apache.tinkerpop.gremlin.structure.strategy.StrategyGraph;
-import org.apache.tinkerpop.gremlin.structure.strategy.StrategyVertex;
-import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Set;
-import java.util.function.Function;
-import java.util.function.Supplier;
-import java.util.function.UnaryOperator;
-
-/**
- * @author Stephen Mallette (http://stephen.genoprime.com)
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class ComputerDataStrategy implements GraphStrategy {
-
- private final Set<String> elementComputeKeys;
-
- public ComputerDataStrategy(final Set<String> elementComputeKeys) {
- this.elementComputeKeys = elementComputeKeys;
- }
-
- @Override
- public <V> UnaryOperator<Function<String[], Iterator<VertexProperty<V>>>> getVertexIteratorsPropertyIteratorStrategy(final StrategyContext<StrategyVertex> ctx, final GraphStrategy strategyComposer) {
- return (f) -> (keys) -> keys.length == 0 ? IteratorUtils.filter(f.apply(keys), property -> !this.elementComputeKeys.contains(property.key())) : f.apply(keys);
- }
-
- @Override
- public <V> UnaryOperator<Function<String[], Iterator<V>>> getVertexIteratorsValueIteratorStrategy(final StrategyContext<StrategyVertex> ctx, final GraphStrategy strategyComposer) {
- return (f) -> (keys) -> IteratorUtils.map(ctx.getCurrent().<V>properties(keys), vertexProperty -> vertexProperty.value());
- }
-
- @Override
- public UnaryOperator<Supplier<Set<String>>> getVertexKeysStrategy(final StrategyContext<StrategyVertex> ctx, final GraphStrategy strategyComposer) {
- return (f) -> () -> IteratorUtils.fill(IteratorUtils.filter(f.get().iterator(), key -> !this.elementComputeKeys.contains(key)), new HashSet<>());
- }
-
- @Override
- public String toString() {
- return StringFactory.graphStrategyString(this);
- }
-
- public static StrategyGraph wrapGraph(final Graph graph, final VertexProgram<?> vertexProgram) {
- return graph.strategy(new ComputerDataStrategy(vertexProgram.getElementComputeKeys()));
- }
-
- public static StrategyVertex wrapVertex(final Vertex vertex, final VertexProgram<?> vertexProgram) {
- return new StrategyVertex(vertex, vertex.graph().strategy(new ComputerDataStrategy(vertexProgram.getElementComputeKeys())));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerGraph.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerGraph.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerGraph.java
new file mode 100644
index 0000000..6168182
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerGraph.java
@@ -0,0 +1,344 @@
+/*
+ * 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.util;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.structure.Direction;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Element;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Property;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.VertexProperty;
+import org.apache.tinkerpop.gremlin.structure.util.ElementHelper;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class ComputerGraph implements Graph {
+
+ private final Graph graph;
+ private final Set<String> computeKeys;
+
+ public ComputerGraph(final Graph graph, final Set<String> elementComputeKeys) {
+ this.graph = graph;
+ this.computeKeys = elementComputeKeys;
+ }
+
+ public static Vertex of(final Vertex vertex, final Set<String> elementComputeKeys) {
+ return new ComputerGraph(vertex.graph(), elementComputeKeys).wrapVertex(vertex);
+ }
+
+ private final Vertex wrapVertex(final Vertex vertex) {
+ return new ComputerVertex(vertex);
+ }
+
+ @Override
+ public Vertex addVertex(final Object... keyValues) {
+ return new ComputerVertex(this.graph.addVertex(keyValues));
+ }
+
+ @Override
+ public <C extends GraphComputer> C compute(final Class<C> graphComputerClass) throws IllegalArgumentException {
+ return this.graph.compute(graphComputerClass);
+ }
+
+ @Override
+ public GraphComputer compute() throws IllegalArgumentException {
+ return this.graph.compute();
+ }
+
+ @Override
+ public Iterator<Vertex> vertices(final Object... vertexIds) {
+ return IteratorUtils.map(this.graph.vertices(vertexIds), vertex -> new ComputerVertex(vertex));
+ }
+
+ @Override
+ public Iterator<Edge> edges(final Object... edgeIds) {
+ return IteratorUtils.map(this.graph.edges(edgeIds), edge -> new ComputerEdge(edge));
+ }
+
+ @Override
+ public Transaction tx() {
+ return this.graph.tx();
+ }
+
+ @Override
+ public Variables variables() {
+ return this.graph.variables();
+ }
+
+ @Override
+ public Configuration configuration() {
+ return this.graph.configuration();
+ }
+
+ @Override
+ public void close() throws Exception {
+ this.graph.close();
+ }
+
+ private class ComputerElement implements Element {
+ private final Element element;
+
+ public ComputerElement(final Element element) {
+ this.element = element;
+ }
+
+ @Override
+ public Object id() {
+ return this.element.id();
+ }
+
+ @Override
+ public String label() {
+ return this.element.label();
+ }
+
+ @Override
+ public Graph graph() {
+ return ComputerGraph.this;
+ }
+
+ @Override
+ public Set<String> keys() {
+ return this.element.keys().stream().filter(key -> !computeKeys.contains(key)).collect(Collectors.toSet());
+ }
+
+ @Override
+ public <V> Property<V> property(final String key) {
+ return new ComputerProperty<>(this.element.property(key));
+ }
+
+ @Override
+ public <V> Property<V> property(final String key, final V value) {
+ return new ComputerProperty<>(this.element.property(key, value));
+ }
+
+ @Override
+ public <V> V value(final String key) throws NoSuchElementException {
+ return this.element.value(key);
+ }
+
+ @Override
+ public void remove() {
+ this.element.remove();
+ }
+
+ @Override
+ public <V> Iterator<? extends Property<V>> properties(final String... propertyKeys) {
+ return (Iterator) IteratorUtils.filter(this.element.properties(propertyKeys), property -> !computeKeys.contains(property.key()));
+ }
+
+ @Override
+ public <V> Iterator<V> values(final String... propertyKeys) {
+ return IteratorUtils.map(this.<V>properties(propertyKeys), property -> property.value());
+ }
+
+ @Override
+ public int hashCode() {
+ return this.element.hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return this.element.toString();
+ }
+
+ @Override
+ public boolean equals(final Object other) {
+ return ElementHelper.areEqual(this, other);
+ }
+
+ protected final Vertex asVertex() {
+ return (Vertex) this.element;
+ }
+
+ protected final Edge asEdge() {
+ return (Edge) this.element;
+ }
+
+ protected final <V> VertexProperty<V> asVertexProperty() {
+ return (VertexProperty<V>) this.element;
+ }
+ }
+
+ ///////////////////////////////////
+
+ private class ComputerVertex extends ComputerElement implements Vertex {
+
+
+ public ComputerVertex(final Vertex vertex) {
+ super(vertex);
+ }
+
+ @Override
+ public <V> VertexProperty<V> property(final String key) {
+ return new ComputerVertexProperty<>(this.asVertex().property(key));
+ }
+
+ @Override
+ public <V> VertexProperty<V> property(final String key, final V value) {
+ return new ComputerVertexProperty<>(this.asVertex().property(key, value));
+ }
+
+ @Override
+ public <V> VertexProperty<V> property(final String key, final V value, final Object... keyValues) {
+ return new ComputerVertexProperty<>(this.asVertex().property(key, value, keyValues));
+ }
+
+ @Override
+ public <V> VertexProperty<V> property(final VertexProperty.Cardinality cardinality, final String key, final V value, final Object... keyValues) {
+ return new ComputerVertexProperty<>(this.asVertex().property(cardinality, key, value, keyValues));
+ }
+
+ @Override
+ public Edge addEdge(final String label, final Vertex inVertex, final Object... keyValues) {
+ return new ComputerEdge(this.asVertex().addEdge(label, inVertex, keyValues));
+ }
+
+ @Override
+ public Iterator<Edge> edges(final Direction direction, final String... edgeLabels) {
+ return IteratorUtils.map(this.asVertex().edges(direction, edgeLabels), edge -> new ComputerEdge(edge));
+ }
+
+ @Override
+ public Iterator<Vertex> vertices(final Direction direction, final String... edgeLabels) {
+ return IteratorUtils.map(this.asVertex().vertices(direction, edgeLabels), vertex -> new ComputerVertex(vertex));
+ }
+
+ @Override
+ public <V> Iterator<VertexProperty<V>> properties(final String... propertyKeys) {
+ return IteratorUtils.map(super.properties(propertyKeys), property -> new ComputerVertexProperty<V>((VertexProperty<V>) property));
+ }
+ }
+
+ ////////////////////////////
+
+ private class ComputerEdge extends ComputerElement implements Edge {
+
+ public ComputerEdge(final Edge edge) {
+ super(edge);
+ }
+
+
+ @Override
+ public Iterator<Vertex> vertices(final Direction direction) {
+ return IteratorUtils.map(this.asEdge().vertices(direction), vertex -> new ComputerVertex(vertex));
+ }
+
+ @Override
+ public Vertex outVertex() {
+ return new ComputerVertex(this.asEdge().outVertex());
+ }
+
+ @Override
+ public Vertex inVertex() {
+ return new ComputerVertex(this.asEdge().inVertex());
+ }
+
+ @Override
+ public <V> Iterator<Property<V>> properties(final String... propertyKeys) {
+ return IteratorUtils.map(super.properties(propertyKeys), property -> new ComputerProperty(property));
+ }
+ }
+
+ ///////////////////////////
+
+ private class ComputerVertexProperty<V> extends ComputerElement implements VertexProperty<V> {
+ public ComputerVertexProperty(final VertexProperty<V> vertexProperty) {
+ super(vertexProperty);
+ }
+
+ @Override
+ public String key() {
+ return this.asVertexProperty().key();
+ }
+
+ @Override
+ public V value() throws NoSuchElementException {
+ return this.<V>asVertexProperty().value();
+ }
+
+ @Override
+ public boolean isPresent() {
+ return this.asVertexProperty().isPresent();
+ }
+
+ @Override
+ public Vertex element() {
+ return new ComputerVertex(this.asVertexProperty().element());
+ }
+
+ @Override
+ public <U> Iterator<Property<U>> properties(final String... propertyKeys) {
+ return IteratorUtils.map(super.properties(propertyKeys), property -> new ComputerProperty(property));
+ }
+ }
+
+ ///////////////////////////
+
+ private class ComputerProperty<V> implements Property<V> {
+
+ private final Property<V> property;
+
+ public ComputerProperty(final Property<V> property) {
+ this.property = property;
+ }
+
+ @Override
+ public String key() {
+ return this.property.key();
+ }
+
+ @Override
+ public V value() throws NoSuchElementException {
+ return this.property.value();
+ }
+
+ @Override
+ public boolean isPresent() {
+ return this.property.isPresent();
+ }
+
+ @Override
+ public Element element() {
+ final Element element = this.property.element();
+ if (element instanceof Vertex)
+ return new ComputerVertex((Vertex) element);
+ else if (element instanceof Edge)
+ return new ComputerEdge((Edge) element);
+ else
+ return new ComputerVertexProperty((VertexProperty) element);
+ }
+
+ @Override
+ public void remove() {
+ this.property.remove();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/GraphComputerHelper.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/GraphComputerHelper.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/GraphComputerHelper.java
index 73f49c9..2b8cf12 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/GraphComputerHelper.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/GraphComputerHelper.java
@@ -18,13 +18,13 @@
*/
package org.apache.tinkerpop.gremlin.process.computer.util;
-import org.apache.tinkerpop.gremlin.process.Traverser;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.structure.Graph;
+import java.io.Serializable;
import java.lang.reflect.Method;
import java.util.Comparator;
import java.util.List;
@@ -71,18 +71,4 @@ public final class GraphComputerHelper {
if (!(b instanceof MapReduce)) return false;
return a.getClass().equals(b.getClass()) && a.getMemoryKey().equals(((MapReduce) b).getMemoryKey());
}
-
- public static <T> Comparator<Traverser<T>> chainComparators(final List<Comparator<T>> comparators) {
- if (comparators.size() == 0) {
- return (a, b) -> a.compareTo(b);
- } else {
- return comparators.stream().map(c -> (Comparator<Traverser<T>>) new Comparator<Traverser<T>>() {
- @Override
- public int compare(final Traverser<T> o1, final Traverser<T> o2) {
- return c.compare(o1.get(), o2.get());
- }
- }).reduce((a, b) -> a.thenComparing(b)).get();
- }
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/map/OrderGlobalStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/map/OrderGlobalStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/map/OrderGlobalStep.java
index 1fcbb11..acd1f9a 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/map/OrderGlobalStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/map/OrderGlobalStep.java
@@ -26,13 +26,16 @@ import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.process.traverser.TraverserRequirement;
import org.apache.tinkerpop.gremlin.process.util.TraverserSet;
import org.apache.tinkerpop.gremlin.structure.Order;
+import org.apache.tinkerpop.gremlin.util.function.ChainedComparator;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Set;
+import java.util.stream.Collectors;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -40,7 +43,6 @@ import java.util.Set;
public final class OrderGlobalStep<S> extends CollectingBarrierStep<S> implements ComparatorHolder<S> {
private final List<Comparator<S>> comparators = new ArrayList<>();
- private Comparator<Traverser<S>> chainedComparator = (Comparator) Order.incr;
public OrderGlobalStep(final Traversal.Admin traversal) {
super(traversal);
@@ -48,18 +50,17 @@ public final class OrderGlobalStep<S> extends CollectingBarrierStep<S> implement
@Override
public void barrierConsumer(final TraverserSet<S> traverserSet) {
- traverserSet.sort(this.chainedComparator);
+ traverserSet.sort(this.comparators.isEmpty() ? new ComparatorTraverser(Order.incr) : new ChainedComparator(ComparatorTraverser.convertComparator((List) this.comparators)));
}
@Override
public void addComparator(final Comparator<S> comparator) {
this.comparators.add(comparator);
- this.chainedComparator = this.comparators.stream().map(c -> (Comparator<Traverser<S>>) new ComparatorTraverser<S>(c)).reduce((a, b) -> a.thenComparing(b)).get();
}
@Override
public List<Comparator<S>> getComparators() {
- return this.comparators;
+ return this.comparators.isEmpty() ? Arrays.asList((Comparator) Order.incr) : this.comparators;
}
@Override
@@ -86,5 +87,9 @@ public final class OrderGlobalStep<S> extends CollectingBarrierStep<S> implement
public int compare(final Traverser<S> traverserA, final Traverser<S> traverserB) {
return this.comparator.compare(traverserA.get(), traverserB.get());
}
+
+ public static <S> List<ComparatorTraverser<S>> convertComparator(final List<Comparator<S>> comparators) {
+ return comparators.stream().map(comparator -> new ComparatorTraverser<>(comparator)).collect(Collectors.toList());
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/util/TraverserSet.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/util/TraverserSet.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/util/TraverserSet.java
index 8a57a84..37fb957 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/util/TraverserSet.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/util/TraverserSet.java
@@ -130,7 +130,7 @@ public class TraverserSet<S> extends AbstractSet<Traverser.Admin<S>> implements
@Override
public String toString() {
- return this.map.values().toString();
+ return this.map.keySet().toString();
}
public void sort(final Comparator<Traverser<S>> comparator) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/function/ChainedComparator.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/function/ChainedComparator.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/function/ChainedComparator.java
new file mode 100644
index 0000000..6068f10
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/function/ChainedComparator.java
@@ -0,0 +1,44 @@
+/*
+ * 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.util.function;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.List;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class ChainedComparator<T> implements Comparator<T>, Serializable {
+
+ private final List<Comparator<T>> comparators;
+ private transient Comparator<T> chain;
+
+ public ChainedComparator(final List<Comparator<T>> comparators) {
+ if (comparators.isEmpty())
+ throw new IllegalArgumentException("A chained comparator requires at least one comparator");
+ this.comparators = comparators;
+ }
+
+ @Override
+ public int compare(final T objectA, final T objectB) {
+ if (null == this.chain) this.chain = this.comparators.stream().reduce((a, b) -> a.thenComparing(b)).get();
+ return this.chain.compare(objectA, objectB);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
index 1f9736f..805400c 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
@@ -21,7 +21,7 @@ package org.apache.tinkerpop.gremlin.process;
import org.apache.tinkerpop.gremlin.AbstractGremlinSuite;
import org.apache.tinkerpop.gremlin.AbstractGremlinTest;
import org.apache.tinkerpop.gremlin.process.computer.ranking.PageRankVertexProgramTest;
-import org.apache.tinkerpop.gremlin.process.computer.util.ComputerDataStrategyTest;
+import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraphTest;
import org.apache.tinkerpop.gremlin.process.graph.traversal.step.branch.BranchTest;
import org.apache.tinkerpop.gremlin.process.graph.traversal.step.branch.ChooseTest;
import org.apache.tinkerpop.gremlin.process.graph.traversal.step.branch.LocalTest;
@@ -67,7 +67,6 @@ import org.apache.tinkerpop.gremlin.process.graph.traversal.step.sideEffect.Side
import org.apache.tinkerpop.gremlin.process.graph.traversal.step.sideEffect.StoreTest;
import org.apache.tinkerpop.gremlin.process.graph.traversal.step.sideEffect.TreeTest;
import org.apache.tinkerpop.gremlin.process.graph.traversal.strategy.TraversalVerificationStrategyTest;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.PartitionStrategyProcessTest;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.ReadOnlyStrategyProcessTest;
import org.junit.runners.model.InitializationError;
import org.junit.runners.model.RunnerBuilder;
@@ -171,7 +170,7 @@ public class ProcessComputerSuite extends AbstractGremlinSuite {
PageRankVertexProgramTest.class,
// strategy
- ComputerDataStrategyTest.class,
+ ComputerGraphTest.class,
// strategy
TraversalVerificationStrategyTest.ComputerTraversals.class,
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerDataStrategyTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerDataStrategyTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerDataStrategyTest.java
deleted file mode 100644
index c9e95b6..0000000
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerDataStrategyTest.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.tinkerpop.gremlin.process.computer.util;
-
-import org.apache.tinkerpop.gremlin.AbstractGremlinTest;
-import org.apache.tinkerpop.gremlin.FeatureRequirement;
-import org.apache.tinkerpop.gremlin.process.TraversalEngine;
-import org.apache.tinkerpop.gremlin.process.UseEngine;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.structure.strategy.StrategyGraph;
-import org.junit.Test;
-
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Set;
-
-import static org.junit.Assert.*;
-
-/**
- * @author Stephen Mallette (http://stephen.genoprime.com)
- */
-@UseEngine(TraversalEngine.Type.COMPUTER)
-public class ComputerDataStrategyTest extends AbstractGremlinTest {
-
- @Test
- @FeatureRequirement(featureClass = Graph.Features.VertexFeatures.class, feature = Graph.Features.VertexFeatures.FEATURE_ADD_VERTICES)
- public void shouldFilterHiddenProperties() {
- final StrategyGraph sg = graph.strategy(new ComputerDataStrategy(new HashSet<>(Arrays.asList("***hidden-guy"))));
-
- final Vertex v = sg.addVertex("***hidden-guy", "X", "not-hidden-guy", "Y");
- final Iterator<VertexProperty<String>> props = v.properties();
- final VertexProperty v1 = props.next();
- assertEquals("Y", v1.value());
- assertEquals("not-hidden-guy", v1.key());
- assertFalse(props.hasNext());
-
- final Iterator<String> values = v.values();
- assertEquals("Y", values.next());
- assertFalse(values.hasNext());
- }
-
- @Test
- @FeatureRequirement(featureClass = Graph.Features.VertexFeatures.class, feature = Graph.Features.VertexFeatures.FEATURE_ADD_VERTICES)
- public void shouldAccessHiddenProperties() {
- final StrategyGraph sg = graph.strategy(new ComputerDataStrategy(new HashSet<>(Arrays.asList("***hidden-guy"))));
-
- final Vertex v = sg.addVertex("***hidden-guy", "X", "not-hidden-guy", "Y");
- final Iterator<VertexProperty<String>> props = v.properties("***hidden-guy");
- final VertexProperty<String> v1 = props.next();
- assertEquals("X", v1.value());
- assertEquals("***hidden-guy", v1.key());
- assertFalse(props.hasNext());
-
- final Iterator<String> values = v.values("***hidden-guy");
- assertEquals("X", values.next());
- assertFalse(values.hasNext());
- }
-
- @Test
- @FeatureRequirement(featureClass = Graph.Features.VertexFeatures.class, feature = Graph.Features.VertexFeatures.FEATURE_ADD_VERTICES)
- public void shouldHideHiddenKeys() {
- final StrategyGraph sg = graph.strategy(new ComputerDataStrategy(new HashSet<>(Arrays.asList("***hidden-guy"))));
-
- final Vertex v = sg.addVertex("***hidden-guy", "X", "not-hidden-guy", "Y");
- final Set<String> keys = v.keys();
- assertTrue(keys.contains("not-hidden-guy"));
- assertFalse(keys.contains("***hidden-guy"));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerGraphTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerGraphTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerGraphTest.java
new file mode 100644
index 0000000..0fd02cc
--- /dev/null
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ComputerGraphTest.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.process.computer.util;
+
+import org.apache.tinkerpop.gremlin.AbstractGremlinTest;
+import org.apache.tinkerpop.gremlin.FeatureRequirement;
+import org.apache.tinkerpop.gremlin.process.TraversalEngine;
+import org.apache.tinkerpop.gremlin.process.UseEngine;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.VertexProperty;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+import static org.junit.Assert.*;
+
+/**
+ * @author Stephen Mallette (http://stephen.genoprime.com)
+ */
+@UseEngine(TraversalEngine.Type.COMPUTER)
+public class ComputerGraphTest extends AbstractGremlinTest {
+
+ @Test
+ @FeatureRequirement(featureClass = Graph.Features.VertexFeatures.class, feature = Graph.Features.VertexFeatures.FEATURE_ADD_VERTICES)
+ public void shouldFilterHiddenProperties() {
+ final ComputerGraph sg = new ComputerGraph(graph, new HashSet<>(Arrays.asList("***hidden-guy")));
+
+ final Vertex v = sg.addVertex("***hidden-guy", "X", "not-hidden-guy", "Y");
+ final Iterator<VertexProperty<String>> props = v.properties();
+ final VertexProperty v1 = props.next();
+ assertEquals("Y", v1.value());
+ assertEquals("not-hidden-guy", v1.key());
+ assertFalse(props.hasNext());
+
+ final Iterator<String> values = v.values();
+ assertEquals("Y", values.next());
+ assertFalse(values.hasNext());
+ }
+
+ @Test
+ @Ignore
+ @FeatureRequirement(featureClass = Graph.Features.VertexFeatures.class, feature = Graph.Features.VertexFeatures.FEATURE_ADD_VERTICES)
+ public void shouldAccessHiddenProperties() {
+ final ComputerGraph sg = new ComputerGraph(graph, new HashSet<>(new HashSet<>(Arrays.asList("***hidden-guy"))));
+
+ final Vertex v = sg.addVertex("***hidden-guy", "X", "not-hidden-guy", "Y");
+ final Iterator<VertexProperty<String>> props = v.properties("***hidden-guy");
+ final VertexProperty<String> v1 = props.next();
+ assertEquals("X", v1.value());
+ assertEquals("***hidden-guy", v1.key());
+ assertFalse(props.hasNext());
+
+ final Iterator<String> values = v.values("***hidden-guy");
+ assertEquals("X", values.next());
+ assertFalse(values.hasNext());
+ }
+
+ @Test
+ @FeatureRequirement(featureClass = Graph.Features.VertexFeatures.class, feature = Graph.Features.VertexFeatures.FEATURE_ADD_VERTICES)
+ public void shouldHideHiddenKeys() {
+ final ComputerGraph sg = new ComputerGraph(graph, new HashSet<>(Arrays.asList("***hidden-guy")));
+
+ final Vertex v = sg.addVertex("***hidden-guy", "X", "not-hidden-guy", "Y");
+ final Set<String> keys = v.keys();
+ assertTrue(keys.contains("not-hidden-guy"));
+ assertFalse(keys.contains("***hidden-guy"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/hadoop-gremlin/conf/hadoop-gryo.properties
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/conf/hadoop-gryo.properties b/hadoop-gremlin/conf/hadoop-gryo.properties
index 7677bd2..8d9de43 100644
--- a/hadoop-gremlin/conf/hadoop-gryo.properties
+++ b/hadoop-gremlin/conf/hadoop-gryo.properties
@@ -30,10 +30,12 @@ gremlin.hadoop.outputLocation=output
giraph.minWorkers=2
giraph.maxWorkers=2
giraph.useOutOfCoreGraph=true
+giraph.useOutOfCoreMessages=true
mapred.map.child.java.opts=-Xmx1024m
mapred.reduce.child.java.opts=-Xmx1024m
giraph.numInputThreads=4
giraph.numComputeThreads=4
+giraph.maxMessagesInMemory=100000
# giraph.vertexOutputFormatThreadSafe=true
# giraph.numOutputThreads=4
# giraph.maxPartitionsInMemory=1
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java
index f23f4a8..1d657b3 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java
@@ -18,7 +18,6 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph;
-import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
import org.apache.giraph.graph.Vertex;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
@@ -26,10 +25,9 @@ import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.util.ComputerDataStrategy;
+import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.structure.strategy.StrategyVertex;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -50,17 +48,12 @@ public final class GiraphComputeVertex extends Vertex<LongWritable, VertexWritab
}
@Override
- public void setConf(final ImmutableClassesGiraphConfiguration<LongWritable, VertexWritable, NullWritable, ObjectWritable> configuration) {
- // do nothing -- no need to store the configuration with the giraph vertex
- }
-
- @Override
public void compute(final Iterable<ObjectWritable> messages) {
final GiraphWorkerContext workerContext = (GiraphWorkerContext) this.getWorkerContext();
- final VertexProgram vertexProgram = workerContext.getVertexProgramPool().take();
+ final VertexProgram<?> vertexProgram = workerContext.getVertexProgramPool().take();
final GiraphMemory memory = workerContext.getMemory();
final GiraphMessenger messenger = workerContext.getMessenger(this, messages);
- final StrategyVertex wrappedVertex = ComputerDataStrategy.wrapVertex(this.getValue().get(), vertexProgram);
+ final org.apache.tinkerpop.gremlin.structure.Vertex wrappedVertex = ComputerGraph.of(this.getValue().get(), vertexProgram.getElementComputeKeys());
///////////
if (!(Boolean) ((RuleWritable) this.getAggregatedValue(Constants.GREMLIN_HADOOP_HALT)).getObject()) {
vertexProgram.execute(wrappedVertex, messenger, memory); // TODO provide a wrapper around TinkerVertex for Edge and non-ComputeKeys manipulation
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java
index 8275ca7..8abaeaa 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java
@@ -48,6 +48,7 @@ import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.ComputerDataStrategy;
+import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
import org.apache.tinkerpop.gremlin.process.computer.util.DefaultComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.util.GraphComputerHelper;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
@@ -75,7 +76,7 @@ public class GiraphGraphComputer extends Configured implements GraphComputer, To
private boolean executed = false;
private final Set<MapReduce> mapReduces = new HashSet<>();
- private VertexProgram vertexProgram;
+ private VertexProgram<?> vertexProgram;
private MapMemory memory = new MapMemory();
public GiraphGraphComputer(final HadoopGraph hadoopGraph) {
@@ -149,7 +150,7 @@ public class GiraphGraphComputer extends Configured implements GraphComputer, To
this.memory.setRuntime(System.currentTimeMillis() - startTime);
final Graph outputGraph = HadoopHelper.getOutputGraph(this.hadoopGraph, null != this.vertexProgram);
- return new DefaultComputerResult(null == this.vertexProgram ? outputGraph : ComputerDataStrategy.wrapGraph(outputGraph, this.vertexProgram), this.memory.asImmutable());
+ return new DefaultComputerResult(null == this.vertexProgram ? outputGraph : new ComputerGraph(outputGraph, this.vertexProgram.getElementComputeKeys()), this.memory.asImmutable());
});
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
index 0dae435..db277ae 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
@@ -40,9 +40,11 @@ import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
import org.apache.tinkerpop.gremlin.process.computer.util.DefaultComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.util.GraphComputerHelper;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertex;
import org.slf4j.Logger;
@@ -191,7 +193,8 @@ public final class SparkGraphComputer implements GraphComputer {
sparkContext.close();
// update runtime and return the newly computed graph
finalMemory.setRuntime(System.currentTimeMillis() - startTime);
- return new DefaultComputerResult(HadoopHelper.getOutputGraph(this.hadoopGraph, null != this.vertexProgram), finalMemory.asImmutable());
+ final Graph outputGraph = HadoopHelper.getOutputGraph(this.hadoopGraph, null != this.vertexProgram);
+ return new DefaultComputerResult(null == this.vertexProgram ? outputGraph : new ComputerGraph(outputGraph, this.vertexProgram.getElementComputeKeys()), finalMemory.asImmutable());
}
}
);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java
index cff5b87..e74f0fd 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java
@@ -40,11 +40,13 @@ import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageCombiner;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import scala.Tuple2;
import java.io.IOException;
import java.util.Optional;
+import java.util.Set;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -59,9 +61,10 @@ public final class SparkHelper {
// execute vertex program
current = current.mapPartitionsToPair(partitionIterator -> { // each partition(Spark)/worker(TP3) has a local copy of the vertex program to reduce object creation
final VertexProgram<M> workerVertexProgram = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration);
+ final Set<String> elementComputeKeys = workerVertexProgram.getElementComputeKeys();
workerVertexProgram.workerIterationStart(memory);
return () -> IteratorUtils.<Tuple2<Object, SparkPayload<M>>, Tuple2<Object, SparkPayload<M>>>map(partitionIterator, keyValue -> {
- workerVertexProgram.execute(keyValue._2().asVertexPayload().getVertex(), keyValue._2().asVertexPayload(), memory);
+ workerVertexProgram.execute(ComputerGraph.of(keyValue._2().asVertexPayload().getVertex(), elementComputeKeys), keyValue._2().asVertexPayload(), memory);
if (!partitionIterator.hasNext()) workerVertexProgram.workerIterationEnd(memory); // is this safe?
return keyValue;
});
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c6fef8f5/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
index d527b67..3f05dbe 100644
--- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
+++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
@@ -23,9 +23,8 @@ import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.util.ComputerDataStrategy;
+import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
import org.apache.tinkerpop.gremlin.process.computer.util.GraphComputerHelper;
-import org.apache.tinkerpop.gremlin.process.traversal.engine.ComputerTraversalEngine;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
@@ -95,9 +94,9 @@ public class TinkerGraphComputer implements GraphComputer {
GraphComputerHelper.validateProgramOnComputer(this, this.vertexProgram);
this.mapReducers.addAll(this.vertexProgram.getMapReducers());
}
-
- final Graph sg = null == this.vertexProgram ? this.graph :
- this.graph.strategy(new ComputerDataStrategy(this.vertexProgram.getElementComputeKeys()));
+ //final Graph computeGraph = this.graph;
+ final Graph computeGraph = null == this.vertexProgram ? this.graph :
+ new ComputerGraph(this.graph, this.vertexProgram.getElementComputeKeys());
this.memory = new TinkerMemory(this.vertexProgram, this.mapReducers);
return CompletableFuture.<ComputerResult>supplyAsync(() -> {
@@ -111,7 +110,7 @@ public class TinkerGraphComputer implements GraphComputer {
this.memory.completeSubRound();
while (true) {
workers.executeVertexProgram(vertexProgram -> vertexProgram.workerIterationStart(this.memory.asImmutable()), vertexProgram);
- final SynchronizedIterator<Vertex> vertices = new SynchronizedIterator<>(sg.vertices());
+ final SynchronizedIterator<Vertex> vertices = new SynchronizedIterator<>(computeGraph.vertices());
workers.executeVertexProgram(vertexProgram -> {
while (true) {
final Vertex vertex = vertices.next();
@@ -137,7 +136,7 @@ public class TinkerGraphComputer implements GraphComputer {
for (final MapReduce mapReduce : mapReducers) {
if (mapReduce.doStage(MapReduce.Stage.MAP)) {
final TinkerMapEmitter<?, ?> mapEmitter = new TinkerMapEmitter<>(mapReduce.doStage(MapReduce.Stage.REDUCE));
- final SynchronizedIterator<Vertex> vertices = new SynchronizedIterator<>(sg.vertices());
+ final SynchronizedIterator<Vertex> vertices = new SynchronizedIterator<>(computeGraph.vertices());
workers.executeMapReduce(workerMapReduce -> {
while (true) {
final Vertex vertex = vertices.next();
@@ -170,7 +169,7 @@ public class TinkerGraphComputer implements GraphComputer {
// update runtime and return the newly computed graph
this.memory.setRuntime(System.currentTimeMillis() - time);
this.memory.complete();
- return new TinkerComputerResult(sg, this.memory.asImmutable());
+ return new TinkerComputerResult(computeGraph, this.memory.asImmutable());
} catch (Exception ex) {
throw new RuntimeException(ex);