You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by dk...@apache.org on 2016/05/27 18:33:50 UTC
[01/20] incubator-tinkerpop git commit: added a VertexTest for g.V(1,
2, blah, 3).name to ensure that 1, 2, 3 are returned. CTR.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1298 430ed4c8f -> 0c366a076
added a VertexTest for g.V(1,2,blah,3).name to ensure that 1,2,3 are returned. CTR.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/3feb7526
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/3feb7526
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/3feb7526
Branch: refs/heads/TINKERPOP-1298
Commit: 3feb7526d32ea07e7874f71480ff8873b9080b08
Parents: 6148e33
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 18 08:13:58 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 18 08:13:58 2016 -0600
----------------------------------------------------------------------
.../traversal/step/map/GroovyVertexTest.groovy | 7 +++++
.../process/traversal/step/map/VertexTest.java | 31 ++++++++++++++++++--
2 files changed, 35 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3feb7526/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyVertexTest.groovy
----------------------------------------------------------------------
diff --git a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyVertexTest.groovy b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyVertexTest.groovy
index 34a567e..7b8974c 100644
--- a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyVertexTest.groovy
+++ b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyVertexTest.groovy
@@ -169,5 +169,12 @@ public abstract class GroovyVertexTest {
public Traversal<Vertex, Vertex> get_g_VX1X_to_XOUT_knowsX(final Object v1Id) {
TraversalScriptHelper.compute("g.V(v1Id).to(Direction.OUT, 'knows')", g, "v1Id", v1Id);
}
+
+ @Override
+ public Traversal<Vertex, String> get_g_VX1_2_3_4X_name(
+ final Object v1Id, final Object v2Id, final Object v3Id, final Object v4Id) {
+ g.V(v3Id).drop().iterate();
+ TraversalScriptHelper.compute("g.V(v1Id, v2Id, v4Id, v3Id).name", g, "v1Id", v1Id, "v2Id", v2Id, "v3Id", v3Id, "v4Id", v4Id);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3feb7526/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/VertexTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/VertexTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/VertexTest.java
index a521027..063af97 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/VertexTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/VertexTest.java
@@ -18,13 +18,17 @@
*/
package org.apache.tinkerpop.gremlin.process.traversal.step.map;
+import org.apache.tinkerpop.gremlin.FeatureRequirement;
import org.apache.tinkerpop.gremlin.LoadGraphWith;
import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest;
import org.apache.tinkerpop.gremlin.process.GremlinProcessRunner;
+import org.apache.tinkerpop.gremlin.process.IgnoreEngine;
import org.apache.tinkerpop.gremlin.process.traversal.P;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalEngine;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -37,7 +41,9 @@ import java.util.Map;
import java.util.Set;
import static org.apache.tinkerpop.gremlin.LoadGraphWith.GraphData.MODERN;
-import static org.junit.Assert.*;
+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)
@@ -103,12 +109,14 @@ public abstract class VertexTest extends AbstractGremlinProcessTest {
public abstract Traversal<Vertex, Vertex> get_g_VX1X_to_XOUT_knowsX(final Object v1Id);
+ public abstract Traversal<Vertex, String> get_g_VX1_2_3_4X_name(final Object v1Id, final Object v2Id, final Object v3Id, final Object v4Id);
+
// GRAPH VERTEX/EDGE
@Test
@LoadGraphWith(MODERN)
public void g_VXlistX1_2_3XX_name() {
- final Traversal<Vertex,String> traversal = get_g_VXlistX1_2_3XX_name();
+ final Traversal<Vertex, String> traversal = get_g_VXlistX1_2_3XX_name();
printTraversalForm(traversal);
checkResults(Arrays.asList("marko", "vadas", "lop"), traversal);
}
@@ -116,7 +124,7 @@ public abstract class VertexTest extends AbstractGremlinProcessTest {
@Test
@LoadGraphWith(MODERN)
public void g_VXlistXv1_v2_v3XX_name() {
- final Traversal<Vertex,String> traversal = get_g_VXlistXv1_v2_v3XX_name();
+ final Traversal<Vertex, String> traversal = get_g_VXlistXv1_v2_v3XX_name();
printTraversalForm(traversal);
checkResults(Arrays.asList("marko", "vadas", "lop"), traversal);
}
@@ -539,6 +547,17 @@ public abstract class VertexTest extends AbstractGremlinProcessTest {
assertFalse(traversal.hasNext());
}
+ @Test
+ @LoadGraphWith
+ @FeatureRequirement(featureClass = Graph.Features.VertexFeatures.class, feature = Graph.Features.VertexFeatures.FEATURE_REMOVE_VERTICES)
+ @IgnoreEngine(TraversalEngine.Type.COMPUTER)
+ public void g_VX1_2_3_4X_name() {
+ final Traversal<Vertex, String> traversal = get_g_VX1_2_3_4X_name(convertToVertexId("marko"), convertToVertexId("vadas"), convertToVertexId("lop"), convertToVertexId("josh"));
+ printTraversalForm(traversal);
+ checkResults(Arrays.asList("marko", "vadas", "josh"), traversal);
+ assertFalse(traversal.hasNext());
+ }
+
public static class Traversals extends VertexTest {
@Override
@@ -677,6 +696,12 @@ public abstract class VertexTest extends AbstractGremlinProcessTest {
}
@Override
+ public Traversal<Vertex, String> get_g_VX1_2_3_4X_name(final Object v1Id, final Object v2Id, final Object v3Id, final Object v4Id) {
+ g.V(v3Id).drop().iterate();
+ return g.V(v1Id, v2Id, v3Id, v4Id).values("name");
+ }
+
+ @Override
public Traversal<Edge, Edge> get_g_EX11X(final Object e11Id) {
return g.E(e11Id);
}
[10/20] incubator-tinkerpop git commit: added a style guide to the
recipes cook book. CTR as the style model was DISCUSS in dev.
Posted by dk...@apache.org.
added a style guide to the recipes cook book. CTR as the style model was DISCUSS in dev.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/66a82cec
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/66a82cec
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/66a82cec
Branch: refs/heads/TINKERPOP-1298
Commit: 66a82cece601af2b04080e6b7a0f4da959a48078
Parents: c61095f
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu May 19 08:12:56 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu May 19 08:12:56 2016 -0600
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
docs/src/recipes/index.asciidoc | 2 +
docs/src/recipes/style-guide.asciidoc | 81 ++++++++++++++++++++++++++++++
3 files changed, 84 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/66a82cec/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 3de3702..628b4f4 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,7 @@ image::https://raw.githubusercontent.com/apache/incubator-tinkerpop/master/docs/
TinkerPop 3.2.1 (NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+* Added a traversal style guide to the recipes cookbook.
* Fixed a bug in master-traversal traverser propagation.
* Added useful methods for custom `VertexPrograms` to be used with `program()`-step.
* Increased the test coverage around traverser propagation within a multi-job OLAP traversal.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/66a82cec/docs/src/recipes/index.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/recipes/index.asciidoc b/docs/src/recipes/index.asciidoc
index fcda27b..8a64818 100644
--- a/docs/src/recipes/index.asciidoc
+++ b/docs/src/recipes/index.asciidoc
@@ -36,6 +36,8 @@ link:http://tinkerpop.apache.org/docs/x.y.z/tutorials/the-gremlin-console/[The G
Traversal Recipes
=================
+include::style-guide.asciidoc[]
+
include::between-vertices.asciidoc[]
include::shortest-path.asciidoc[]
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/66a82cec/docs/src/recipes/style-guide.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/recipes/style-guide.asciidoc b/docs/src/recipes/style-guide.asciidoc
new file mode 100644
index 0000000..14cbad9
--- /dev/null
+++ b/docs/src/recipes/style-guide.asciidoc
@@ -0,0 +1,81 @@
+////
+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.
+////
+[[style-guide]]
+Style Guide
+-----------
+
+Gremlin is a data flow language where each new step concatenation alters the stream accordingly. This aspect of the
+language allows users to easily "build-up" a traversal (literally) step-by-step until the expected results are
+returned. For instance:
+
+[gremlin-groovy,modern]
+----
+g.V(1)
+g.V(1).out('knows')
+g.V(1).out('knows').out('created')
+g.V(1).out('knows').out('created').groupCount()
+g.V(1).out('knows').out('created').groupCount().by('name')
+----
+
+A drawback of building up a traversal is that users tend to create long, single line traversal that are hard to read.
+For simple traversals, a single line is fine. For complex traversals, there are few formatting patterns that should be followed
+which will yield cleaner, easier to understand traversals. For instance, the last traversal above would be written:
+
+[gremlin-groovy,modern]
+----
+g.V(1).out('knows').out('created').
+ groupCount().by('name')
+----
+
+Lets look at a complex traversal and analyze each line according to the recommended formatting rule is subscribes to.
+
+[gremlin-groovy,modern]
+----
+g.V().out('knows').out('created'). <1>
+ group().by('lang').by(). <2>
+ select('java').unfold(). <3>
+ in('created').hasLabel('person'). <4>
+ order(). <5>
+ by(inE().count(),decr). <6>
+ by('age',incr).
+ dedup().limit(10).values('name') <7>
+----
+
+
+<1> A sequence of `ins().outs().filters().etc()` on a single line until it gets too long.
+<2> When a barrier (reducer, aggregator, etc.) is used, put it on a new line.
+<3> When a next line component is an "add on" to the previous line component, 2 space indent.
+The `select()`-step in this context is "almost like" a `by()`-modulator as its projecting data out of the `group()`.
+The `unfold()`-step is a data formatting necessity that should not be made too prominent.
+<4> Back to a series of `ins().outs().filters().etc()` on a single line.
+<5> `order()` is a barrier step and thus, should be on a new line.
+<6> If there is only one `by()`-modulator (or a series of short ones), keep it on one line, else each `by()` is a new line.
+<7> Back to a series `ins().outs().filters().etc()`.
+
+Style Guide Rules
+~~~~~~~~~~~~~~~~~
+
+A generalization of the specifics above are presented below.
+
+* Always use 2 space indent.
+* No newline should ever have the same indent as the line starting with the traversal source `g`.
+* Barrier steps should form line breaks unless they are simple (e.g. `sum()`).
+* Complex `by()`-modulators form indented "paragraphs."
+* Standard filters, maps, flatMaps remain on the same line until they get too long.
+
+Given the diversity of traversals and the complexities introduced by lambdas (for example), these rules will not always
+lead to optimal representations. However, by in large, the style rules above will help make 90% of traversals look great.
[20/20] incubator-tinkerpop git commit: Don't let users mess with
halted traversers. Instead simply provide all preformatted values.
Posted by dk...@apache.org.
Don't let users mess with halted traversers. Instead simply provide all preformatted values.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/0c366a07
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/0c366a07
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/0c366a07
Branch: refs/heads/TINKERPOP-1298
Commit: 0c366a0762874b8848d5cbe26501960bb3c3feb0
Parents: e13221e
Author: Daniel Kuppitz <da...@hotmail.com>
Authored: Fri May 27 20:32:27 2016 +0200
Committer: Daniel Kuppitz <da...@hotmail.com>
Committed: Fri May 27 20:32:27 2016 +0200
----------------------------------------------------------------------
data/script-csv-export.groovy | 15 +++-----
.../bulkdumping/BulkExportVertexProgram.java | 38 ++++++++++++++++----
2 files changed, 36 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/0c366a07/data/script-csv-export.groovy
----------------------------------------------------------------------
diff --git a/data/script-csv-export.groovy b/data/script-csv-export.groovy
index 7a6da22..4e6d5b1 100644
--- a/data/script-csv-export.groovy
+++ b/data/script-csv-export.groovy
@@ -23,21 +23,14 @@ import org.apache.tinkerpop.gremlin.process.computer.bulkdumping.BulkExportVerte
def stringify(vertex) {
def result = null
- def haltedTraversers = vertex.property(TraversalVertexProgram.HALTED_TRAVERSERS)
- if (haltedTraversers.isPresent()) {
- def properties = vertex.value(BulkExportVertexProgram.BULK_EXPORT_PROPERTIES).split("\1")*.split("\2", 2)*.toList()
+ def rows = vertex.property(BulkExportVertexProgram.BULK_EXPORT_PROPERTIES)
+ if (rows.isPresent()) {
def writer = new StringWriter()
def w = new CSVWriter(writer)
- haltedTraversers.value().each { def t ->
- def values = []
- properties.each { def property, def format ->
- def value = t.path(property)
- values << (format.isEmpty() ? value.toString() : String.format(format, value))
- }
- w.writeNext((String[]) values, false)
+ rows.value().each { def row ->
+ w.writeNext((String[]) row, false)
}
result = writer.toString().trim()
- writer.close()
}
return result
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/0c366a07/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkExportVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkExportVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkExportVertexProgram.java
index cdd6b28..8f65c11 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkExportVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkExportVertexProgram.java
@@ -29,6 +29,8 @@ import org.apache.tinkerpop.gremlin.process.computer.VertexComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
+import org.apache.tinkerpop.gremlin.process.traversal.Path;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
@@ -37,9 +39,13 @@ import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.javatuples.Tuple;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
/**
* @author Daniel Kuppitz (http://gremlin.guru)
@@ -50,7 +56,8 @@ public class BulkExportVertexProgram implements VertexProgram<Tuple> {
public static final String BULK_EXPORT_PROPERTIES = String.join(".", BULK_EXPORT_VERTEX_PROGRAM_CFG_PREFIX, "properties");
private Configuration configuration;
- private String properties;
+ private Map<String, String> properties;
+ private List<String> sortedProperties;
private Set<VertexComputeKey> vertexComputeKeys;
private BulkExportVertexProgram() {
@@ -62,7 +69,13 @@ public class BulkExportVertexProgram implements VertexProgram<Tuple> {
if (config != null) {
ConfigurationUtils.copy(config, configuration);
}
- properties = configuration.getString(BULK_EXPORT_PROPERTIES);
+ properties = new HashMap<>();
+ sortedProperties = new ArrayList<>();
+ for (final String tuple : configuration.getString(BULK_EXPORT_PROPERTIES, "").split("\1")) {
+ final String[] parts = tuple.split("\2", -1);
+ properties.put(parts[0], parts[1]);
+ sortedProperties.add(parts[0]);
+ }
vertexComputeKeys = Collections.singleton(VertexComputeKey.of(BULK_EXPORT_PROPERTIES, false));
}
@@ -81,9 +94,23 @@ public class BulkExportVertexProgram implements VertexProgram<Tuple> {
@Override
public void execute(final Vertex sourceVertex, final Messenger<Tuple> messenger, final Memory memory) {
final VertexProperty<TraverserSet> haltedTraversers = sourceVertex.property(TraversalVertexProgram.HALTED_TRAVERSERS);
- if (haltedTraversers.isPresent()) {
- sourceVertex.property(BULK_EXPORT_PROPERTIES, properties);
- }
+ haltedTraversers.ifPresent(traverserSet -> {
+ final List<List<String>> rows = new ArrayList<>();
+ for (final Traverser t : (Iterable<Traverser>) traverserSet) {
+ final List<String> columns = new ArrayList<>();
+ final Path path = t.path();
+ final Iterable<String> keys = properties.isEmpty()
+ ? t.path().labels().stream().flatMap(Collection::stream).sorted().collect(Collectors.toSet())
+ : sortedProperties;
+ for (final String key : keys) {
+ final String format = properties.getOrDefault(key, "");
+ final Object value = path.get(key);
+ columns.add("".equals(format) ? value.toString() : String.format(format, value));
+ }
+ rows.add(columns);
+ }
+ sourceVertex.property(BULK_EXPORT_PROPERTIES, rows);
+ });
}
@Override
@@ -139,7 +166,6 @@ public class BulkExportVertexProgram implements VertexProgram<Tuple> {
@Override
public BulkExportVertexProgram create(final Graph graph) {
configuration.setProperty(BULK_EXPORT_PROPERTIES, String.join("\1", properties));
- graph.configuration().setProperty("gremlin.hadoop.graphWriter", "org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptOutputFormat");
return (BulkExportVertexProgram) VertexProgram.createVertexProgram(graph, configuration);
}
[16/20] incubator-tinkerpop git commit: Merge remote-tracking branch
'origin/tp31'
Posted by dk...@apache.org.
Merge remote-tracking branch 'origin/tp31'
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/44d40f60
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/44d40f60
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/44d40f60
Branch: refs/heads/TINKERPOP-1298
Commit: 44d40f60664b8b758f02bc8e04300b931b2c6622
Parents: 6ffeaf1 aa447eb
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Mon May 23 08:15:09 2016 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Mon May 23 08:15:09 2016 -0400
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../apache/tinkerpop/gremlin/server/op/session/Session.java | 6 +++++-
2 files changed, 6 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/44d40f60/CHANGELOG.asciidoc
----------------------------------------------------------------------
[02/20] incubator-tinkerpop git commit: upmerged a test to ensure
g.V(1, 2, blah, 3) returns 1, 2, 3 vertices. CTR.
Posted by dk...@apache.org.
upmerged a test to ensure g.V(1,2,blah,3) returns 1, 2, 3 vertices. CTR.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/c61095f6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/c61095f6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/c61095f6
Branch: refs/heads/TINKERPOP-1298
Commit: c61095f6b37e1d6514c91f14e09dea0ef2e99f07
Parents: 4f03ea1 3feb752
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 18 08:32:42 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 18 08:32:42 2016 -0600
----------------------------------------------------------------------
.../traversal/step/map/GroovyVertexTest.groovy | 13 ++++++++--
.../process/traversal/step/map/VertexTest.java | 27 +++++++++++++++++++-
2 files changed, 37 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c61095f6/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyVertexTest.groovy
----------------------------------------------------------------------
diff --cc gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyVertexTest.groovy
index 451739f,7b8974c..3c4f8b4
--- a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyVertexTest.groovy
+++ b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyVertexTest.groovy
@@@ -31,13 -31,13 +31,15 @@@ public abstract class GroovyVertexTest
public static class Traversals extends VertexTest {
@Override
- public Traversal<Vertex, String> get_g_VXlistXv1_v2_v3XX_name(final Vertex v1, final Vertex v2, final Vertex v3) {
- public Traversal<Vertex, String> get_g_VXlistXv1_v2_v3XX_name() {
- TraversalScriptHelper.compute("g.V(ids).name", g, "ids", [convertToVertex(graph, "marko"), convertToVertex(graph, "vadas"), convertToVertex(graph, "lop")])
++ public Traversal<Vertex, String> get_g_VXlistXv1_v2_v3XX_name(
++ final Vertex v1, final Vertex v2, final Vertex v3) {
+ new ScriptTraversal<>(g, "gremlin-groovy", "g.V(ids).name", "ids", [v1, v2, v3])
}
@Override
- public Traversal<Vertex, String> get_g_VXlistX1_2_3XX_name(final Object v1Id, final Object v2Id, final Object v3Id) {
- public Traversal<Vertex, String> get_g_VXlistX1_2_3XX_name() {
- TraversalScriptHelper.compute("g.V(ids).name", g, "ids", [convertToVertexId(graph, "marko"), convertToVertexId(graph, "vadas"), convertToVertexId(graph, "lop")])
++ public Traversal<Vertex, String> get_g_VXlistX1_2_3XX_name(
++ final Object v1Id, final Object v2Id, final Object v3Id) {
+ new ScriptTraversal<>(g, "gremlin-groovy", "g.V(ids).name", "ids", [v1Id, v2Id, v3Id])
}
@Override
@@@ -167,7 -167,14 +169,14 @@@
@Override
public Traversal<Vertex, Vertex> get_g_VX1X_to_XOUT_knowsX(final Object v1Id) {
- TraversalScriptHelper.compute("g.V(v1Id).to(Direction.OUT, 'knows')", g, "v1Id", v1Id);
+ new ScriptTraversal<>(g, "gremlin-groovy", "g.V(v1Id).to(Direction.OUT, 'knows')", "v1Id", v1Id)
}
+
+ @Override
+ public Traversal<Vertex, String> get_g_VX1_2_3_4X_name(
+ final Object v1Id, final Object v2Id, final Object v3Id, final Object v4Id) {
+ g.V(v3Id).drop().iterate();
- TraversalScriptHelper.compute("g.V(v1Id, v2Id, v4Id, v3Id).name", g, "v1Id", v1Id, "v2Id", v2Id, "v3Id", v3Id, "v4Id", v4Id);
++ new ScriptTraversal<>(g, "gremlin-groovy", "g.V(v1Id, v2Id, v4Id, v3Id).name", "v1Id", v1Id, "v2Id", v2Id, "v3Id", v3Id, "v4Id", v4Id)
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/c61095f6/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/VertexTest.java
----------------------------------------------------------------------
[06/20] incubator-tinkerpop git commit: minor tweaks to
TraverserVertexProgram to make it more efficient.
Posted by dk...@apache.org.
minor tweaks to TraverserVertexProgram to make it more efficient.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/7d29ef32
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/7d29ef32
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/7d29ef32
Branch: refs/heads/TINKERPOP-1298
Commit: 7d29ef32a42b143f9654457a46ab64a57446ef14
Parents: a6aacdc
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 18 17:12:40 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 18 17:12:40 2016 -0600
----------------------------------------------------------------------
.../process/computer/traversal/TraversalVertexProgram.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/7d29ef32/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
index aba44d3..c7e7ef9 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
@@ -304,8 +304,9 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
return false;
} else {
// finalize locally with any last traversers dangling in the local traversal
- while (this.traversal.get().getEndStep().hasNext()) {
- final Traverser.Admin traverser = this.traversal.get().getEndStep().next();
+ final Step<?, ?> endStep = this.traversal.get().getEndStep();
+ while (endStep.hasNext()) {
+ final Traverser.Admin traverser = endStep.next();
traverser.detach();
haltedTraversers.add(traverser);
}
[17/20] incubator-tinkerpop git commit: added BulkExportVertexProgram
Posted by dk...@apache.org.
added BulkExportVertexProgram
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/8ad5b62d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/8ad5b62d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/8ad5b62d
Branch: refs/heads/TINKERPOP-1298
Commit: 8ad5b62de0e2c538be069a9b9cc7db07c9d30e0c
Parents: 44d40f6
Author: Daniel Kuppitz <da...@hotmail.com>
Authored: Sun May 22 09:37:41 2016 +0200
Committer: Daniel Kuppitz <da...@hotmail.com>
Committed: Tue May 24 19:51:21 2016 +0200
----------------------------------------------------------------------
.../bulkdumping/BulkExportVertexProgram.java | 163 +++++++++++++++++++
.../gremlin/structure/io/gryo/GryoMapper.java | 4 +-
2 files changed, 166 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8ad5b62d/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkExportVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkExportVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkExportVertexProgram.java
new file mode 100644
index 0000000..cdd6b28
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkdumping/BulkExportVertexProgram.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.process.computer.bulkdumping;
+
+import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationUtils;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
+import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
+import org.apache.tinkerpop.gremlin.process.computer.Messenger;
+import org.apache.tinkerpop.gremlin.process.computer.VertexComputeKey;
+import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
+import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.VertexProperty;
+import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
+import org.javatuples.Tuple;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * @author Daniel Kuppitz (http://gremlin.guru)
+ */
+public class BulkExportVertexProgram implements VertexProgram<Tuple> {
+
+ public static final String BULK_EXPORT_VERTEX_PROGRAM_CFG_PREFIX = "gremlin.bulkExportVertexProgram";
+ public static final String BULK_EXPORT_PROPERTIES = String.join(".", BULK_EXPORT_VERTEX_PROGRAM_CFG_PREFIX, "properties");
+
+ private Configuration configuration;
+ private String properties;
+ private Set<VertexComputeKey> vertexComputeKeys;
+
+ private BulkExportVertexProgram() {
+ }
+
+ @Override
+ public void loadState(final Graph graph, final Configuration config) {
+ configuration = new BaseConfiguration();
+ if (config != null) {
+ ConfigurationUtils.copy(config, configuration);
+ }
+ properties = configuration.getString(BULK_EXPORT_PROPERTIES);
+ vertexComputeKeys = Collections.singleton(VertexComputeKey.of(BULK_EXPORT_PROPERTIES, false));
+ }
+
+ @Override
+ public void storeState(final Configuration config) {
+ VertexProgram.super.storeState(config);
+ if (configuration != null) {
+ ConfigurationUtils.copy(configuration, config);
+ }
+ }
+
+ @Override
+ public void setup(final Memory memory) {
+ }
+
+ @Override
+ public void execute(final Vertex sourceVertex, final Messenger<Tuple> messenger, final Memory memory) {
+ final VertexProperty<TraverserSet> haltedTraversers = sourceVertex.property(TraversalVertexProgram.HALTED_TRAVERSERS);
+ if (haltedTraversers.isPresent()) {
+ sourceVertex.property(BULK_EXPORT_PROPERTIES, properties);
+ }
+ }
+
+ @Override
+ public boolean terminate(final Memory memory) {
+ return properties == null || properties.isEmpty() || !memory.isInitialIteration();
+ }
+
+ @Override
+ public Set<MessageScope> getMessageScopes(final Memory memory) {
+ return Collections.emptySet();
+ }
+
+ @SuppressWarnings({"CloneDoesntDeclareCloneNotSupportedException", "CloneDoesntCallSuperClone"})
+ @Override
+ public VertexProgram<Tuple> clone() {
+ return this;
+ }
+
+ @Override
+ public GraphComputer.ResultGraph getPreferredResultGraph() {
+ return GraphComputer.ResultGraph.NEW;
+ }
+
+ @Override
+ public GraphComputer.Persist getPreferredPersist() {
+ return GraphComputer.Persist.EDGES;
+ }
+
+ @Override
+ public Set<VertexComputeKey> getVertexComputeKeys() {
+ return this.vertexComputeKeys;
+ }
+
+ @Override
+ public String toString() {
+ return StringFactory.vertexProgramString(this);
+ }
+
+ public static Builder build() {
+ return new Builder();
+ }
+
+ public static class Builder extends AbstractVertexProgramBuilder<Builder> {
+
+ private List<String> properties;
+
+ private Builder() {
+ super(BulkExportVertexProgram.class);
+ properties = new ArrayList<>();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public BulkExportVertexProgram create(final Graph graph) {
+ configuration.setProperty(BULK_EXPORT_PROPERTIES, String.join("\1", properties));
+ graph.configuration().setProperty("gremlin.hadoop.graphWriter", "org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptOutputFormat");
+ return (BulkExportVertexProgram) VertexProgram.createVertexProgram(graph, configuration);
+ }
+
+ public Builder key(final String key, final String format) {
+ properties.add(key + "\2" + format);
+ return this;
+ }
+
+ public Builder key(final String key) {
+ this.key(key, "");
+ return this;
+ }
+
+ public Builder keys(final String... keys) {
+ for (final String key : keys) {
+ this.key(key, "");
+ }
+ return this;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8ad5b62d/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java
index d510706..083317a 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java
@@ -94,6 +94,7 @@ import java.time.Year;
import java.time.YearMonth;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
+import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
@@ -292,6 +293,7 @@ public final class GryoMapper implements Mapper<Kryo> {
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(UUID.class, kryo -> new UUIDSerializer(), 17));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(URI.class, kryo -> new URISerializer(), 72));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(VertexTerminator.class, null, 13));
+ add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(AbstractMap.SimpleEntry.class, kryo -> new EntrySerializer(), 120)); // ***LAST ID***
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ReferenceEdge.class, null, 81));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ReferenceVertexProperty.class, null, 82));
@@ -353,7 +355,7 @@ public final class GryoMapper implements Mapper<Kryo> {
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(GroupStepV3d0.GroupBiOperatorV3d0.class, null, 113));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(RangeGlobalStep.RangeBiOperator.class, null, 114));
add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(OrderGlobalStep.OrderBiOperator.class, kryo -> new JavaSerializer(), 118)); // because they contain traversals
- add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ProfileStep.ProfileBiOperator.class, null, 119)); // ***LAST ID***
+ add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ProfileStep.ProfileBiOperator.class, null, 119));
}};
private final List<IoRegistry> registries = new ArrayList<>();
[13/20] incubator-tinkerpop git commit: Merge branch 'TINKERPOP-1305'
Posted by dk...@apache.org.
Merge branch 'TINKERPOP-1305'
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/9c53c73e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/9c53c73e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/9c53c73e
Branch: refs/heads/TINKERPOP-1298
Commit: 9c53c73e6475d2c701833648a18eaff029626087
Parents: 3c613d9 5bcf0b0
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri May 20 22:08:41 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri May 20 22:08:41 2016 -0600
----------------------------------------------------------------------
docs/src/reference/the-traversal.asciidoc | 30 ++-
.../process/computer/GiraphGraphComputer.java | 6 +
.../gremlin/process/computer/ProgramPhase.java | 40 ++++
.../peerpressure/PeerPressureVertexProgram.java | 21 +-
.../traversal/MemoryTraversalSideEffects.java | 28 +--
.../computer/traversal/SingleMessenger.java | 49 -----
.../traversal/TraversalVertexProgram.java | 78 +++++--
.../computer/traversal/TraverserExecutor.java | 203 -----------------
.../computer/traversal/WorkerExecutor.java | 220 +++++++++++++++++++
.../traversal/step/VertexComputing.java | 6 +-
.../step/map/PageRankVertexProgramStep.java | 5 +-
.../step/map/PeerPressureVertexProgramStep.java | 13 +-
.../step/map/ProgramVertexProgramStep.java | 6 +-
.../step/map/TraversalVertexProgramStep.java | 10 +-
.../traversal/step/map/VertexProgramStep.java | 21 +-
.../optimization/GraphFilterStrategy.java | 3 +-
.../process/computer/util/EmptyMemory.java | 91 ++++++++
.../process/computer/util/SingleMessenger.java | 49 +++++
.../traversal/step/util/ImmutablePath.java | 3 +-
.../step/map/GroovyPageRankTest.groovy | 5 +
.../step/map/GroovyPeerPressureTest.groovy | 5 +
.../traversal/step/map/PageRankTest.java | 26 +++
.../traversal/step/map/PeerPressureTest.java | 31 +++
.../process/traversal/step/map/ProgramTest.java | 49 +++--
.../process/computer/SparkGraphComputer.java | 7 +-
.../optimization/SparkInterceptorStrategy.java | 3 +-
.../SparkSingleIterationStrategy.java | 3 +-
.../SparkStarBarrierInterceptor.java | 3 +-
28 files changed, 651 insertions(+), 363 deletions(-)
----------------------------------------------------------------------
[15/20] incubator-tinkerpop git commit: Named the thread pool for
gremlin server sessions. CTR
Posted by dk...@apache.org.
Named the thread pool for gremlin server sessions. CTR
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/aa447eb0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/aa447eb0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/aa447eb0
Branch: refs/heads/TINKERPOP-1298
Commit: aa447eb0390625d561e018315d8719c666ad2d31
Parents: 3feb752
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Mon May 23 08:14:24 2016 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Mon May 23 08:14:24 2016 -0400
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../apache/tinkerpop/gremlin/server/op/session/Session.java | 6 +++++-
2 files changed, 6 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/aa447eb0/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 1e33187..5bf3ada 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,7 @@ image::https://raw.githubusercontent.com/apache/incubator-tinkerpop/master/docs/
TinkerPop 3.1.3 (NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+* Named the thread pool used by Gremlin Server sessions: "gremlin-server-session-$n".
* Fixed a bug in `BulkSet.equals()` which made itself apparent when using `store()` and `aggregate()` with labeled `cap()`.
* Ensured that all asserts of vertex and edge counts were being applied properly in the test suite.
* Fixed bug in `gremlin-driver` where certain channel-level errors would not allow the driver to reconnect.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/aa447eb0/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
index 01e7bbd..0ed2041 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
@@ -23,6 +23,7 @@ import org.apache.tinkerpop.gremlin.server.Context;
import org.apache.tinkerpop.gremlin.server.GraphManager;
import org.apache.tinkerpop.gremlin.server.Settings;
import org.apache.tinkerpop.gremlin.server.util.LifeCycleHook;
+import org.apache.tinkerpop.gremlin.server.util.ThreadFactoryUtil;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,6 +36,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -59,11 +61,13 @@ public class Session {
*/
private final GremlinExecutor gremlinExecutor;
+ private final ThreadFactory threadFactoryWorker = ThreadFactoryUtil.create("session-%d");
+
/**
* By binding the session to run ScriptEngine evaluations in a specific thread, each request will respect
* the ThreadLocal nature of Graph implementations.
*/
- private final ExecutorService executor = Executors.newSingleThreadExecutor();
+ private final ExecutorService executor = Executors.newSingleThreadExecutor(threadFactoryWorker);
private final ConcurrentHashMap<String, Session> sessions;
[12/20] incubator-tinkerpop git commit: Did some more indent work on
the recipes - formatting only CTR
Posted by dk...@apache.org.
Did some more indent work on the recipes - formatting only CTR
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/3c613d9f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/3c613d9f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/3c613d9f
Branch: refs/heads/TINKERPOP-1298
Commit: 3c613d9f8777c30b73f6c5aed687a4642d7d0a75
Parents: 62f0e2d
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Thu May 19 16:49:47 2016 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Thu May 19 16:49:47 2016 -0400
----------------------------------------------------------------------
docs/src/recipes/between-vertices.asciidoc | 14 +++++++++-----
docs/src/recipes/if-then-based-grouping.asciidoc | 13 +++++++------
docs/src/recipes/index.asciidoc | 6 ++++--
docs/src/recipes/shortest-path.asciidoc | 19 ++++++++++++-------
4 files changed, 32 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3c613d9f/docs/src/recipes/between-vertices.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/recipes/between-vertices.asciidoc b/docs/src/recipes/between-vertices.asciidoc
index e8f9845..03a4b3c 100644
--- a/docs/src/recipes/between-vertices.asciidoc
+++ b/docs/src/recipes/between-vertices.asciidoc
@@ -47,7 +47,7 @@ two vertices.
The basic pattern of using `where()` step to find the "other" known vertex can be applied in far more complex
scenarios. Consider the following schema:
-image:recipe-job-schema.png[width=500]
+image:recipe-job-schema.png[width=750]
Assume that the goal is to find information about a known job and a known person. Specifically, the idea would be
to extract the known job, the company that created the job, the date it was created by the company and whether or not
@@ -83,14 +83,18 @@ g.V(vRexsterJob1).as('job').
inE('created').as('created').
outV().as('company').
select('job').
- coalesce(__.in('appliesTo').where(__.in('completes').is(vStephen)), constant(false)).as('application').
- select('job', 'company', 'created', 'application').by().by().by('creationDate').by()
+ coalesce(__.in('appliesTo').where(__.in('completes').is(vStephen)),
+ constant(false)).as('application').
+ select('job', 'company', 'created', 'application').
+ by().by().by('creationDate').by()
g.V(vRexsterJob1, vBlueprintsJob1).as('job').
inE('created').as('created').
outV().as('company').
select('job').
- coalesce(__.in('appliesTo').where(__.in('completes').is(vBob)), constant(false)).as('application').
- select('job', 'company', 'created', 'application').by().by().by('creationDate').by()
+ coalesce(__.in('appliesTo').where(__.in('completes').is(vBob)),
+ constant(false)).as('application').
+ select('job', 'company', 'created', 'application').
+ by().by().by('creationDate').by()
----
While the traversals above are more complex, the pattern for finding "things" between two vertices is largely the same.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3c613d9f/docs/src/recipes/if-then-based-grouping.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/recipes/if-then-based-grouping.asciidoc b/docs/src/recipes/if-then-based-grouping.asciidoc
index fc319b7..dc0025f 100644
--- a/docs/src/recipes/if-then-based-grouping.asciidoc
+++ b/docs/src/recipes/if-then-based-grouping.asciidoc
@@ -35,8 +35,8 @@ a domain concept such as "young", "old" and "very old".
g.V().hasLabel("person").groupCount().by(values("age").choose(
is(lt(28)),constant("young"),
choose(is(lt(30)),
- constant("old"),
- constant("very old"))))
+ constant("old"),
+ constant("very old"))))
----
Note that the `by` modulator has been altered from simply taking a string key of "age" to take a `Traversal`. That
@@ -61,10 +61,11 @@ there is another option to consider with `coalesce`:
[gremlin-groovy,modern]
----
-g.V().hasLabel("person").groupCount().by(values("age").coalesce(
- is(lt(28)).constant("young"),
- is(lt(30)).constant("old"),
- constant("very old")))
+g.V().hasLabel("person").
+ groupCount().by(values("age").
+ coalesce(is(lt(28)).constant("young"),
+ is(lt(30)).constant("old"),
+ constant("very old")))
----
The answer is the same, but this traversal removes the nested `choose`, which makes it easier to read.
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3c613d9f/docs/src/recipes/index.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/recipes/index.asciidoc b/docs/src/recipes/index.asciidoc
index 8a64818..89d4660 100644
--- a/docs/src/recipes/index.asciidoc
+++ b/docs/src/recipes/index.asciidoc
@@ -36,17 +36,19 @@ link:http://tinkerpop.apache.org/docs/x.y.z/tutorials/the-gremlin-console/[The G
Traversal Recipes
=================
-include::style-guide.asciidoc[]
-
include::between-vertices.asciidoc[]
include::shortest-path.asciidoc[]
include::if-then-based-grouping.asciidoc[]
+include::cycle-detection.asciidoc[]
+
Implementation Recipes
======================
+include::style-guide.asciidoc[]
+
include::traversal-component-reuse.asciidoc[]
[[contributing]]
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3c613d9f/docs/src/recipes/shortest-path.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/recipes/shortest-path.asciidoc b/docs/src/recipes/shortest-path.asciidoc
index 161646a..3f380f1 100644
--- a/docs/src/recipes/shortest-path.asciidoc
+++ b/docs/src/recipes/shortest-path.asciidoc
@@ -40,7 +40,8 @@ v3.addEdge("knows", v4)
g = graph.traversal()
g.V(1).repeat(out().simplePath()).until(hasId(5)).path().limit(1) <1>
g.V(1).repeat(out().simplePath()).until(hasId(5)).path().count(local) <2>
-g.V(1).repeat(out().simplePath()).until(hasId(5)).path().group().by(count(local)).next() <3>
+g.V(1).repeat(out().simplePath()).until(hasId(5)).path().
+ group().by(count(local)).next() <3>
----
<1> The traversal starts at vertex with the identifier of "1" and repeatedly traverses on out edges "until" it finds a
@@ -69,15 +70,19 @@ v4.addEdge("knows", v5, "weight", 0.25)
v2.addEdge("knows", v3, "weight", 0.25)
v3.addEdge("knows", v4, "weight", 0.25)
g = graph.traversal()
-g.V(1).repeat(out().simplePath()).until(hasId(5)).path().group().by(count(local)).next() <1>
-g.V(1).repeat(outE().inV().simplePath()).until(hasId(5)). <2>
- path().by(coalesce(values('weight'), constant(0.0))).map(unfold().sum())
+g.V(1).repeat(out().simplePath()).until(hasId(5)).path().
+ group().by(count(local)).next() <1>
+g.V(1).repeat(outE().inV().simplePath()).until(hasId(5)).
+ path().by(coalesce(values('weight'),
+ constant(0.0))).
+ map(unfold().sum()) <2>
g.V(1).repeat(outE().inV().simplePath()).until(hasId(5)).
path().by(constant(0.0)).by('weight').map(unfold().sum()) <3>
-g.V(1).repeat(outE().inV().simplePath()).until(hasId(5)). <4>
+g.V(1).repeat(outE().inV().simplePath()).until(hasId(5)).
path().as('p').
- map(unfold().coalesce(values('weight'),constant(0.0)).sum()).as('cost').
- select('cost','p')
+ map(unfold().coalesce(values('weight'),
+ constant(0.0)).sum()).as('cost').
+ select('cost','p') <4>
----
<1> Note that the shortest path as determined by the structure of the graph is the same.
[19/20] incubator-tinkerpop git commit: Merge branch 'TINKERPOP-1298'
of https://git-wip-us.apache.org/repos/asf/incubator-tinkerpop into
TINKERPOP-1298
Posted by dk...@apache.org.
Merge branch 'TINKERPOP-1298' of https://git-wip-us.apache.org/repos/asf/incubator-tinkerpop into TINKERPOP-1298
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/e13221ea
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/e13221ea
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/e13221ea
Branch: refs/heads/TINKERPOP-1298
Commit: e13221ea1e5761ba48e5ae604473ab4e8f6ade5c
Parents: e13f91a 430ed4c
Author: Daniel Kuppitz <da...@hotmail.com>
Authored: Fri May 27 18:48:02 2016 +0200
Committer: Daniel Kuppitz <da...@hotmail.com>
Committed: Fri May 27 18:48:02 2016 +0200
----------------------------------------------------------------------
----------------------------------------------------------------------
[11/20] incubator-tinkerpop git commit: Added "cycle detection"
recipe. CTR
Posted by dk...@apache.org.
Added "cycle detection" recipe. CTR
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/62f0e2d2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/62f0e2d2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/62f0e2d2
Branch: refs/heads/TINKERPOP-1298
Commit: 62f0e2d2da911520898f1f2204bbf413ba023368
Parents: 66a82ce
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Thu May 19 16:49:30 2016 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Thu May 19 16:49:30 2016 -0400
----------------------------------------------------------------------
docs/src/recipes/cycle-detection.asciidoc | 61 +++++++++++++++++++++++++
docs/static/images/graph-cycle.png | Bin 0 -> 774901 bytes
2 files changed, 61 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/62f0e2d2/docs/src/recipes/cycle-detection.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/recipes/cycle-detection.asciidoc b/docs/src/recipes/cycle-detection.asciidoc
new file mode 100644
index 0000000..864b760
--- /dev/null
+++ b/docs/src/recipes/cycle-detection.asciidoc
@@ -0,0 +1,61 @@
+////
+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.
+////
+[[cycle-detection]]
+Cycle Detection
+---------------
+
+A cycle occurs in a graph where a path loops back on itself to the originating vertex. For example, in the graph
+depticted below Gremlin could be use to detect the cycle among vertices `A-B-C`.
+
+image:graph-cycle.png[width=250]
+
+[gremlin-groovy]
+----
+vA = graph.addVertex(id, 'a')
+vB = graph.addVertex(id, 'b')
+vC = graph.addVertex(id, 'c')
+vD = graph.addVertex(id, 'd')
+vA.addEdge("knows", vB)
+vB.addEdge("knows", vC)
+vC.addEdge("knows", vA)
+vA.addEdge("knows", vD)
+vC.addEdge("knows", vD)
+g.V().as("a").repeat(out().simplePath()).times(2).
+ where(out().as("a")).path() <1>
+g.V().as("a").repeat(out().simplePath()).times(2).
+ where(out().as("a")).path().
+ dedup().by(unfold().order().by(id).dedup().fold()) <2>
+----
+
+<1> Gremlin starts its traversal from a vertex labeled "a" and traverses `out()` from each vertex filtering on the
+`simplePath`, which removes paths with repeated objects. The steps going `out()` are repeated twice as in this case
+the length of the cycle is known to be three and there is no need to exceed that. The traversal filters with a
+`where()` to see only return paths that end with where it started at "a".
+<2> The previous query returned the `A-B-C` cycle, but it returned three paths which were all technically the same
+cycle. It returned three, because there was one for each vertex that started the cycle (i.e. one for `A`, one for `B`
+and one for `C`). This next line introduce deduplication to only return unique cycles.
+
+The above case assumed that the need was to only detect cycles over a path length of three. It also respected the
+directionality of the edges by only considering outgoing ones. What would need to change to detect cycles of
+arbitrary length over both incoming and outgoing edges in the modern graph?
+
+[gremlin-groovy,modern]
+----
+g.V().as("a").repeat(both().simplePath()).emit(loops().is(gt(1))).
+ both().where(eq("a")).path().
+ dedup().by(unfold().order().by(id).dedup().fold())
+----
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/62f0e2d2/docs/static/images/graph-cycle.png
----------------------------------------------------------------------
diff --git a/docs/static/images/graph-cycle.png b/docs/static/images/graph-cycle.png
new file mode 100644
index 0000000..967e0ad
Binary files /dev/null and b/docs/static/images/graph-cycle.png differ
[08/20] incubator-tinkerpop git commit: added EmptyMemory and its
used as the memory for the first generateProgram() of an OLAP chain. Much
better than Optional(Memory).
Posted by dk...@apache.org.
added EmptyMemory and its used as the memory for the first generateProgram() of an OLAP chain. Much better than Optional(Memory).
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/b100f033
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/b100f033
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/b100f033
Branch: refs/heads/TINKERPOP-1298
Commit: b100f033e02ee5647dd31b2d556a1dcadac29818
Parents: 6107ff9
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 18 19:36:18 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 18 19:36:18 2016 -0600
----------------------------------------------------------------------
.../traversal/step/VertexComputing.java | 8 ++--
.../step/map/PageRankVertexProgramStep.java | 4 +-
.../step/map/PeerPressureVertexProgramStep.java | 9 +----
.../step/map/ProgramVertexProgramStep.java | 10 ++---
.../step/map/TraversalVertexProgramStep.java | 9 ++---
.../traversal/step/map/VertexProgramStep.java | 6 +--
.../optimization/GraphFilterStrategy.java | 4 +-
.../process/computer/util/EmptyMemory.java | 39 ++++++++++++++++++++
.../optimization/SparkInterceptorStrategy.java | 6 +--
.../SparkSingleIterationStrategy.java | 5 +--
10 files changed, 59 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
index b599c42..33f185a 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
@@ -25,8 +25,6 @@ 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.util.Optional;
-
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
@@ -51,11 +49,11 @@ public interface VertexComputing {
/**
* Generate the {@link VertexProgram}.
*
- * @param graph the {@link Graph} that the program will be executed over.
- * @param memoryOptional a reference to a {@link Memory} from the previous OLAP job if it exists.
+ * @param graph the {@link Graph} that the program will be executed over.
+ * @param memory the {@link Memory} from the previous OLAP job if it exists, else its an empty memory structure.
* @return the generated vertex program instance.
*/
- public VertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional);
+ public VertexProgram generateProgram(final Graph graph, final Memory memory);
/**
* @deprecated As of release 3.2.1. Please use {@link VertexComputing#getComputer()}.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
index 660f5b8..fcb2eec 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
@@ -21,7 +21,6 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal.step.map;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.lambda.HaltedTraversersCountTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies;
@@ -38,7 +37,6 @@ import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.util.Collections;
import java.util.List;
-import java.util.Optional;
import java.util.Set;
/**
@@ -84,7 +82,7 @@ public final class PageRankVertexProgramStep extends VertexProgramStep implement
}
@Override
- public PageRankVertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional) {
+ public PageRankVertexProgram generateProgram(final Graph graph, final Memory memory) {
final Traversal.Admin<Vertex, Edge> detachedTraversal = this.edgeTraversal.getPure();
detachedTraversal.setStrategies(TraversalStrategies.GlobalCache.getStrategies(graph.getClass()));
final PageRankVertexProgram.Builder builder = PageRankVertexProgram.build()
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
index 2ae36a5..7bd726e 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
@@ -36,7 +36,6 @@ import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.util.Collections;
import java.util.List;
-import java.util.Optional;
import java.util.Set;
/**
@@ -85,15 +84,9 @@ public final class PeerPressureVertexProgramStep extends VertexProgramStep imple
}
@Override
- public PeerPressureVertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional) {
+ public PeerPressureVertexProgram generateProgram(final Graph graph, final Memory memory) {
final Traversal.Admin<Vertex, Edge> detachedTraversal = this.edgeTraversal.getPure();
detachedTraversal.setStrategies(TraversalStrategies.GlobalCache.getStrategies(graph.getClass()));
- /*
- memoryOptional.ifPresent(memory -> {
- if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
- builder.configure(TraversalVertexProgram.HALTED_TRAVERSERS, memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
- });
- */
return PeerPressureVertexProgram.build()
.property(this.clusterProperty)
.maxIterations(this.times)
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
index d060c99..82e70dd 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
@@ -31,7 +31,6 @@ import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.util.HashMap;
import java.util.Map;
-import java.util.Optional;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -51,16 +50,13 @@ public final class ProgramVertexProgramStep extends VertexProgramStep {
}
@Override
- public VertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional) {
+ public VertexProgram generateProgram(final Graph graph, final Memory memory) {
final MapConfiguration base = new MapConfiguration(this.configuration);
base.setDelimiterParsingDisabled(true);
PureTraversal.storeState(base, ROOT_TRAVERSAL, TraversalHelper.getRootTraversal(this.getTraversal()).clone());
base.setProperty(STEP_ID, this.getId());
- memoryOptional.ifPresent(memory -> {
- if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS)) {
- TraversalVertexProgram.storeHaltedTraversers(base, memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
- }
- });
+ if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
+ TraversalVertexProgram.storeHaltedTraversers(base, memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
return VertexProgram.createVertexProgram(graph, base);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
index 5851f35..58e44a2 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
@@ -33,7 +33,6 @@ import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.util.Collections;
import java.util.List;
-import java.util.Optional;
import java.util.Set;
/**
@@ -64,17 +63,15 @@ public final class TraversalVertexProgramStep extends VertexProgramStep implemen
}
@Override
- public TraversalVertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional) {
+ public TraversalVertexProgram generateProgram(final Graph graph, final Memory memory) {
final Traversal.Admin<?, ?> computerSpecificTraversal = this.computerTraversal.getPure();
computerSpecificTraversal.setStrategies(TraversalStrategies.GlobalCache.getStrategies(graph.getClass()).clone());
this.getTraversal().getStrategies().toList().forEach(computerSpecificTraversal.getStrategies()::addStrategies);
computerSpecificTraversal.setSideEffects(new MemoryTraversalSideEffects(this.getTraversal().getSideEffects()));
computerSpecificTraversal.setParent(this);
final TraversalVertexProgram.Builder builder = TraversalVertexProgram.build().traversal(computerSpecificTraversal);
- memoryOptional.ifPresent(memory -> {
- if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
- builder.haltedTraversers(memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
- });
+ if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
+ builder.haltedTraversers(memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
return builder.create(graph);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
index ac3de30..a5d7105 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
@@ -25,6 +25,7 @@ import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.VertexComputing;
+import org.apache.tinkerpop.gremlin.process.computer.util.EmptyMemory;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
@@ -36,7 +37,6 @@ import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedE
import org.apache.tinkerpop.gremlin.structure.Graph;
import java.util.NoSuchElementException;
-import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -64,7 +64,7 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com
if (this.first && this.getPreviousStep() instanceof EmptyStep) {
this.first = false;
final Graph graph = this.getTraversal().getGraph().get();
- future = this.generateComputer(graph).program(this.generateProgram(graph, Optional.empty())).submit();
+ future = this.generateComputer(graph).program(this.generateProgram(graph, EmptyMemory.instance())).submit();
final ComputerResult result = future.get();
this.processMemorySideEffects(result.memory());
return this.getTraversal().getTraverserGenerator().generate(result, this, 1l);
@@ -72,7 +72,7 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com
final Traverser.Admin<ComputerResult> traverser = this.starts.next();
final Graph graph = traverser.get().graph();
final Memory memory = traverser.get().memory();
- future = this.generateComputer(graph).program(this.generateProgram(graph, Optional.of(memory))).submit();
+ future = this.generateComputer(graph).program(this.generateProgram(graph, memory)).submit();
final ComputerResult result = future.get();
this.processMemorySideEffects(result.memory());
return traverser.split(result, this);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java
index e4cf5ff..c32777b 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java
@@ -23,6 +23,7 @@ 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.computer.traversal.step.map.VertexProgramStep;
+import org.apache.tinkerpop.gremlin.process.computer.util.EmptyMemory;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
@@ -41,7 +42,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
/**
@@ -60,7 +60,7 @@ public final class GraphFilterStrategy extends AbstractTraversalStrategy<Travers
return;
final Graph graph = traversal.getGraph().orElse(EmptyGraph.instance()); // given that this strategy only works for single OLAP jobs, the graph is the traversal graph
for (final TraversalVertexProgramStep step : TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal)) { // will be zero or one step
- final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, Optional.empty()).getTraversal().get().clone();
+ final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, EmptyMemory.instance()).getTraversal().get().clone();
if (!computerTraversal.isLocked())
computerTraversal.applyStrategies();
final Computer computer = step.getComputer();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/EmptyMemory.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/EmptyMemory.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/EmptyMemory.java
new file mode 100644
index 0000000..72b1bbf
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/EmptyMemory.java
@@ -0,0 +1,39 @@
+/*
+ * 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.Memory;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class EmptyMemory {
+
+ private static final Memory INSTANCE = new ImmutableMemory(new MapMemory());
+
+ private EmptyMemory() {
+
+ }
+
+ public static Memory instance() {
+ return INSTANCE;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/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
index a2cc518..bb8476a 100644
--- 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
@@ -20,8 +20,8 @@
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.Memory;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
+import org.apache.tinkerpop.gremlin.process.computer.util.EmptyMemory;
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;
@@ -30,8 +30,6 @@ import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.op
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
-import java.util.Optional;
-
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
@@ -46,7 +44,7 @@ public final class SparkInterceptorStrategy extends AbstractTraversalStrategy<Tr
public void apply(final Traversal.Admin<?, ?> traversal) {
final Graph graph = traversal.getGraph().orElse(EmptyGraph.instance()); // best guess at what the graph will be as its dynamically determined
for (final TraversalVertexProgramStep step : TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal)) {
- final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, Optional.empty()).getTraversal().get().clone();
+ final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, EmptyMemory.instance()).getTraversal().get().clone();
if (!computerTraversal.isLocked())
computerTraversal.applyStrategies();
if (SparkStarBarrierInterceptor.isLegal(computerTraversal)) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b100f033/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java
index 280c7ba..a4acf4c 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java
@@ -20,8 +20,8 @@
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.Memory;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
+import org.apache.tinkerpop.gremlin.process.computer.util.EmptyMemory;
import org.apache.tinkerpop.gremlin.process.traversal.Scope;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
@@ -37,7 +37,6 @@ import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
import java.util.Arrays;
import java.util.HashSet;
-import java.util.Optional;
import java.util.Set;
/**
@@ -61,7 +60,7 @@ public final class SparkSingleIterationStrategy extends AbstractTraversalStrateg
public void apply(final Traversal.Admin<?, ?> traversal) {
final Graph graph = traversal.getGraph().orElse(EmptyGraph.instance()); // best guess at what the graph will be as its dynamically determined
for (final TraversalVertexProgramStep step : TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal)) {
- final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, Optional.empty()).getTraversal().get().clone();
+ final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, EmptyMemory.instance()).getTraversal().get().clone();
if (!computerTraversal.isLocked())
computerTraversal.applyStrategies();
boolean doesMessagePass = TraversalHelper.hasStepOfAssignableClassRecursively(Scope.global, MULTI_ITERATION_CLASSES, computerTraversal);
[07/20] incubator-tinkerpop git commit: made a PageRank test a bit
more robust. running full integration over night.
Posted by dk...@apache.org.
made a PageRank test a bit more robust. running full integration 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/6107ff95
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/6107ff95
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/6107ff95
Branch: refs/heads/TINKERPOP-1298
Commit: 6107ff95e47129d6266a68410c01c17b04af95cf
Parents: 7d29ef3
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 18 19:24:57 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 18 19:24:57 2016 -0600
----------------------------------------------------------------------
.../process/traversal/step/map/GroovyPageRankTest.groovy | 2 +-
.../gremlin/process/traversal/step/map/PageRankTest.java | 10 ++++++----
2 files changed, 7 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6107ff95/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
----------------------------------------------------------------------
diff --git a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
index 8608c4d..13787cc 100644
--- a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
+++ b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
@@ -71,7 +71,7 @@ public abstract class GroovyPageRankTest {
}
@Override
- public Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX() {
+ public Traversal<Vertex, Map<String, List<Object>>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX() {
new ScriptTraversal<>(g, "gremlin-groovy", "g.V.out('created').pageRank().by(bothE()).by('projectRank').times(0).valueMap('name','projectRank')")
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6107ff95/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
index 1e18b48..57a3b3f 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
@@ -49,7 +49,7 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
public abstract Traversal<Vertex, Vertex> get_g_V_pageRank();
- public abstract Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX();
+ public abstract Traversal<Vertex, Map<String, List<Object>>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX();
public abstract Traversal<Vertex, String> get_g_V_pageRank_order_byXpageRank_decrX_name();
@@ -82,10 +82,12 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
@Test
@LoadGraphWith(MODERN)
public void g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX() {
- final Traversal<Vertex, Map<String, Object>> traversal = get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX();
+ final Traversal<Vertex, Map<String, List<Object>>> traversal = get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX();
printTraversalForm(traversal);
+ final List<Map<String, List<Object>>> result = traversal.toList();
+ assertEquals(4, result.size());
final Map<String, Double> map = new HashMap<>();
- traversal.forEachRemaining(m -> map.put(((List<String>) m.get("name")).get(0), ((List<Double>) m.get("projectRank")).get(0)));
+ result.forEach(m -> map.put((String) m.get("name").get(0), (Double) m.get("projectRank").get(0)));
assertEquals(2, map.size());
assertTrue(map.containsKey("lop"));
assertTrue(map.containsKey("ripple"));
@@ -244,7 +246,7 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
}
@Override
- public Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX() {
+ public Traversal<Vertex, Map<String, List<Object>>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX() {
return g.V().out("created").pageRank().by(__.bothE()).by("projectRank").times(0).valueMap("name", "projectRank");
}
[14/20] incubator-tinkerpop git commit: updated CHANGELOG and upgrade
docs.
Posted by dk...@apache.org.
updated CHANGELOG and upgrade docs.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/6ffeaf1b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/6ffeaf1b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/6ffeaf1b
Branch: refs/heads/TINKERPOP-1298
Commit: 6ffeaf1bf95dbc3403475e3fb9801d2a7870de2b
Parents: 9c53c73
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri May 20 22:11:55 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri May 20 22:11:55 2016 -0600
----------------------------------------------------------------------
CHANGELOG.asciidoc | 6 ++++++
docs/src/upgrade/release-3.2.x-incubating.asciidoc | 9 +++++++++
2 files changed, 15 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6ffeaf1b/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 628b4f4..eb79ae8 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,12 @@ image::https://raw.githubusercontent.com/apache/incubator-tinkerpop/master/docs/
TinkerPop 3.2.1 (NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+* Fixed a severe bug around halted traversers in a multi-job OLAP traversal chain.
+* Ensure a separation of `GraphComputer` and `VertexProgram` configurations in `SparkGraphComputer` and `GiraphGraphComputer`.
+* `PeerPressureVertexProgram` now supports dynamic initial vote strength calculations.
+* Added `EmptyMemory` for ease of use when no memory exists.
+* Updated `VertexComputing.generateProgram()` API to include `Memory`. (*breaking*)
+* `ImmutablePath.TailPath` is now serializable like `ImmutablePath`.
* Added a traversal style guide to the recipes cookbook.
* Fixed a bug in master-traversal traverser propagation.
* Added useful methods for custom `VertexPrograms` to be used with `program()`-step.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6ffeaf1b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/upgrade/release-3.2.x-incubating.asciidoc b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
index 1b5aa51..66e5f8c 100644
--- a/docs/src/upgrade/release-3.2.x-incubating.asciidoc
+++ b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
@@ -112,6 +112,15 @@ Upgrading for Providers
Graph System Providers
^^^^^^^^^^^^^^^^^^^^^^
+VertexComputing API Change
++++++++++++++++++++++++
+
+The `VertexComputing` API is used by steps that wrap a `VertexProgram`. There is a method called
+`VertexComputing.generateProgram()` that has changed which now takes a second argument of `Memory`. To upgrade, simply
+fix the method signature of your `VertexComputing` implementations. The `Memory` argument can be safely ignored to
+effect the exact same semantics as prior. However, now previous OLAP job `Memory` can be leveraged when constructing
+the next `VertexProgram` in an OLAP traversal chain.
+
Interrupting Traversals
+++++++++++++++++++++++
[18/20] incubator-tinkerpop git commit: Added script and
configuration file that can be used for OLAP CSV exports.
Posted by dk...@apache.org.
Added script and configuration file that can be used for OLAP CSV exports.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/e13f91aa
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/e13f91aa
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/e13f91aa
Branch: refs/heads/TINKERPOP-1298
Commit: e13f91aacddd4cb5722ac3a6e100fdf70b0c33fd
Parents: 8ad5b62
Author: Daniel Kuppitz <da...@hotmail.com>
Authored: Mon May 23 18:24:22 2016 +0200
Committer: Daniel Kuppitz <da...@hotmail.com>
Committed: Tue May 24 19:51:22 2016 +0200
----------------------------------------------------------------------
data/script-csv-export.groovy | 43 +++++++++++++++
.../conf/hadoop-csv-export.properties | 56 ++++++++++++++++++++
2 files changed, 99 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e13f91aa/data/script-csv-export.groovy
----------------------------------------------------------------------
diff --git a/data/script-csv-export.groovy b/data/script-csv-export.groovy
new file mode 100644
index 0000000..7a6da22
--- /dev/null
+++ b/data/script-csv-export.groovy
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+@Grab(group = 'com.opencsv', module = 'opencsv', version = '3.7')
+import com.opencsv.*
+
+import org.apache.tinkerpop.gremlin.process.computer.bulkdumping.BulkExportVertexProgram
+
+def stringify(vertex) {
+ def result = null
+ def haltedTraversers = vertex.property(TraversalVertexProgram.HALTED_TRAVERSERS)
+ if (haltedTraversers.isPresent()) {
+ def properties = vertex.value(BulkExportVertexProgram.BULK_EXPORT_PROPERTIES).split("\1")*.split("\2", 2)*.toList()
+ def writer = new StringWriter()
+ def w = new CSVWriter(writer)
+ haltedTraversers.value().each { def t ->
+ def values = []
+ properties.each { def property, def format ->
+ def value = t.path(property)
+ values << (format.isEmpty() ? value.toString() : String.format(format, value))
+ }
+ w.writeNext((String[]) values, false)
+ }
+ result = writer.toString().trim()
+ writer.close()
+ }
+ return result
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e13f91aa/hadoop-gremlin/conf/hadoop-csv-export.properties
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/conf/hadoop-csv-export.properties b/hadoop-gremlin/conf/hadoop-csv-export.properties
new file mode 100644
index 0000000..3e1f8da
--- /dev/null
+++ b/hadoop-gremlin/conf/hadoop-csv-export.properties
@@ -0,0 +1,56 @@
+# 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.
+gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph
+gremlin.hadoop.graphReader=org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat
+gremlin.hadoop.graphWriter=org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptOutputFormat
+gremlin.hadoop.jarsInDistributedCache=true
+gremlin.hadoop.defaultGraphComputer=org.apache.tinkerpop.gremlin.spark.process.computer.SparkGraphComputer
+
+gremlin.hadoop.inputLocation=output
+gremlin.hadoop.scriptOutputFormat.script=script-csv-export.groovy
+gremlin.hadoop.outputLocation=export
+
+####################################
+# SparkGraphComputer Configuration #
+####################################
+spark.master=local[4]
+spark.executor.memory=1g
+spark.serializer=org.apache.tinkerpop.gremlin.spark.structure.io.gryo.GryoSerializer
+# spark.kryo.registrationRequired=true
+# spark.storage.memoryFraction=0.2
+# spark.eventLog.enabled=true
+# spark.eventLog.dir=/tmp/spark-event-logs
+# spark.ui.killEnabled=true
+
+#####################################
+# GiraphGraphComputer Configuration #
+#####################################
+giraph.minWorkers=2
+giraph.maxWorkers=2
+giraph.useOutOfCoreGraph=true
+giraph.useOutOfCoreMessages=true
+mapreduce.map.java.opts=-Xmx1024m
+mapreduce.reduce.java.opts=-Xmx1024m
+giraph.numInputThreads=2
+giraph.numComputeThreads=2
+# giraph.maxPartitionsInMemory=1
+# giraph.userPartitionCount=2
+## MapReduce of GiraphGraphComputer ##
+# mapreduce.job.maps=2
+# mapreduce.job.reduces=1
+
+
[04/20] incubator-tinkerpop git commit: TraverserExecutor is smart to
not do another iteration if the traversers will simply be returned to the
master traversals as output. Updated the PageRankTest with a better test.
Posted by dk...@apache.org.
TraverserExecutor is smart to not do another iteration if the traversers will simply be returned to the master traversals as output. Updated the PageRankTest with a better test.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/07e12995
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/07e12995
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/07e12995
Branch: refs/heads/TINKERPOP-1298
Commit: 07e12995ae6bed0a3b1da3942bc54371d94a66a9
Parents: 4ee3028
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 18 17:02:22 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 18 17:02:22 2016 -0600
----------------------------------------------------------------------
.../process/computer/traversal/TraverserExecutor.java | 10 ++++++++--
.../process/traversal/step/map/GroovyPageRankTest.groovy | 4 ++--
.../gremlin/process/traversal/step/map/PageRankTest.java | 10 ++++++----
3 files changed, 16 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/07e12995/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
index f5facdb..c5f9bb5 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
@@ -159,7 +159,10 @@ public final class TraverserExecutor {
final TraverserSet<Object> barrierSet = barrier.nextBarrier();
IteratorUtils.removeOnNext(barrierSet.iterator()).forEachRemaining(traverser -> {
traverser.addLabels(step.getLabels()); // this might need to be generalized for working with global barriers too
- if (traverser.isHalted() && ((!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) || getHostingVertex(traverser.get()).equals(vertex))) {
+ if (traverser.isHalted() &&
+ (returnHaltedTraversers ||
+ (!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
+ getHostingVertex(traverser.get()).equals(vertex))) {
traverser.detach();
if (returnHaltedTraversers)
memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
@@ -181,7 +184,10 @@ public final class TraverserExecutor {
}
} else { // LOCAL PROCESSING
step.forEachRemaining(traverser -> {
- if (traverser.isHalted() && ((!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) || getHostingVertex(traverser.get()).equals(vertex))) {
+ if (traverser.isHalted() &&
+ (returnHaltedTraversers ||
+ (!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
+ getHostingVertex(traverser.get()).equals(vertex))) {
traverser.detach();
if (returnHaltedTraversers)
memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/07e12995/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
----------------------------------------------------------------------
diff --git a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
index e27e8c0..8608c4d 100644
--- a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
+++ b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
@@ -71,8 +71,8 @@ public abstract class GroovyPageRankTest {
}
@Override
- public Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX() {
- new ScriptTraversal<>(g, "gremlin-groovy", "g.V.out('created').pageRank().by(bothE()).by('projectRank').valueMap('name','projectRank')")
+ public Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX() {
+ new ScriptTraversal<>(g, "gremlin-groovy", "g.V.out('created').pageRank().by(bothE()).by('projectRank').times(0).valueMap('name','projectRank')")
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/07e12995/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
index 5a3ed9b..1e18b48 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
@@ -49,7 +49,7 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
public abstract Traversal<Vertex, Vertex> get_g_V_pageRank();
- public abstract Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX();
+ public abstract Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX();
public abstract Traversal<Vertex, String> get_g_V_pageRank_order_byXpageRank_decrX_name();
@@ -82,7 +82,7 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
@Test
@LoadGraphWith(MODERN)
public void g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX() {
- final Traversal<Vertex, Map<String, Object>> traversal = get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX();
+ final Traversal<Vertex, Map<String, Object>> traversal = get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX();
printTraversalForm(traversal);
final Map<String, Double> map = new HashMap<>();
traversal.forEachRemaining(m -> map.put(((List<String>) m.get("name")).get(0), ((List<Double>) m.get("projectRank")).get(0)));
@@ -90,6 +90,8 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
assertTrue(map.containsKey("lop"));
assertTrue(map.containsKey("ripple"));
assertTrue(map.get("lop") > map.get("ripple"));
+ assertEquals(3.0d, map.get("lop"), 0.001d);
+ assertEquals(1.0d, map.get("ripple"), 0.001d);
assertFalse(traversal.hasNext());
}
@@ -242,8 +244,8 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
}
@Override
- public Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX() {
- return g.V().out("created").pageRank().by(__.bothE()).by("projectRank").valueMap("name", "projectRank");
+ public Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_timesX0X_valueMapXname_projectRankX() {
+ return g.V().out("created").pageRank().by(__.bothE()).by("projectRank").times(0).valueMap("name", "projectRank");
}
@Override
[03/20] incubator-tinkerpop git commit: fixed HALTED_TRAVERSER bug
where traversers were not at their 'resting location'. This only shows up in
multi-OLAP job chains. HALTED_TRAVERESERS (master travesal) are now
propagated via Configuration and NOT via s
Posted by dk...@apache.org.
fixed HALTED_TRAVERSER bug where traversers were not at their 'resting location'. This only shows up in multi-OLAP job chains. HALTED_TRAVERESERS (master travesal) are now propagated via Configuration and NOT via sideEffects. This is more elegant and there are now helper methods in TraversalVertexProgram to make getting master halted traversers easy. Updated the-travesal.asciidoc to reflect how to use ProgramStep with it. Had to change a method signature in VertexComputing to account for this... its slight (breaking), but no one is using this right now, I'm sure of it. And if they are, the chaneg is trivial, just add a new argument to a method signature and ignore it to make it be like how it is in 3.2.0. ImmutablePath.TailPath was not serializable and this caused issues for HALTED_TRAVERSERS as well. Everything else fixed up.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/4ee30289
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/4ee30289
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/4ee30289
Branch: refs/heads/TINKERPOP-1298
Commit: 4ee302899da37a0ad0ebab6bc4954ed39b69d6d9
Parents: c61095f
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 18 16:15:40 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 18 16:15:40 2016 -0600
----------------------------------------------------------------------
docs/src/reference/the-traversal.asciidoc | 23 +++++++--
.../process/computer/GiraphGraphComputer.java | 6 +++
.../traversal/MemoryTraversalSideEffects.java | 1 -
.../traversal/TraversalVertexProgram.java | 50 +++++++++++++++++---
.../computer/traversal/TraverserExecutor.java | 21 +++++---
.../traversal/step/VertexComputing.java | 8 +++-
.../step/map/PageRankVertexProgramStep.java | 7 +--
.../step/map/PeerPressureVertexProgramStep.java | 11 ++++-
.../step/map/ProgramVertexProgramStep.java | 10 +++-
.../step/map/TraversalVertexProgramStep.java | 13 +++--
.../traversal/step/map/VertexProgramStep.java | 21 +++-----
.../optimization/GraphFilterStrategy.java | 3 +-
.../traversal/step/util/ImmutablePath.java | 3 +-
.../step/map/GroovyPageRankTest.groovy | 5 ++
.../traversal/step/map/PageRankTest.java | 22 +++++++++
.../process/traversal/step/map/ProgramTest.java | 38 ++++++++-------
.../process/computer/SparkGraphComputer.java | 7 ++-
.../optimization/SparkInterceptorStrategy.java | 5 +-
.../SparkSingleIterationStrategy.java | 4 +-
19 files changed, 185 insertions(+), 73 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/docs/src/reference/the-traversal.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/reference/the-traversal.asciidoc b/docs/src/reference/the-traversal.asciidoc
index 71708e2..3394e90 100644
--- a/docs/src/reference/the-traversal.asciidoc
+++ b/docs/src/reference/the-traversal.asciidoc
@@ -1436,6 +1436,8 @@ are provided below.
[source,java]
----
+private TraverserSet<Object> haltedTraversers;
+
public void loadState(final Graph graph, final Configuration configuration) {
VertexProgram.super.loadState(graph, configuration);
this.traversal = PureTraversal.loadState(configuration, VertexProgramStep.ROOT_TRAVERSAL, graph);
@@ -1444,18 +1446,29 @@ public void loadState(final Graph graph, final Configuration configuration) {
this.memoryComputeKeys.addAll(MemoryTraversalSideEffects.getMemoryComputeKeys(this.traversal.get()));
// if master-traversal traversers may be propagated, create a memory compute key
this.memoryComputeKeys.add(MemoryComputeKey.of(TraversalVertexProgram.HALTED_TRAVERSERS, Operator.addAll, false, false));
+ // returns an empty traverser set if there are no halted traversers
+ this.haltedTraversers = TraversalVertexProgram.getHaltedTraversers(configuration);
+}
+
+public void storeState(final Configuration configuration) {
+ VertexProgram.super.storeState(configuration);
+ // if halted traversers is null or empty, it does nothing
+ TraversalVertexProgram.storeHaltedTraversers(configuration, this.haltedTraversers);
}
public void setup(final Memory memory) {
- if(sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS)) {
- // haltedTraversers in setup() represent master-traversal traversers
- // for example, from a traversal of the form g.V().groupCount().program(...)
- TraverserSet<Object> haltedTraversers = sideEffects.get(TraversalVertexProgram.HALTED_TRAVERSERS);
- // do as you please with this information
+ if(null != this.haltedTraversers) {
+ // do what you like with the halted master traversal traversers
}
+ // once used, no need to keep that information around (master)
+ if(null != this.haltedTraversers)
+ this.haltedTraversers.clear()
}
public void execute(final Vertex vertex, final Messenger messenger, final Memory memory) {
+ // once used, no need to keep that information around (workers)
+ if(null != this.haltedTraversers)
+ this.haltedTraversers.clear()
if(vertex.property(TraversalVertexProgram.HALTED_TRAVERSERS).isPresent()) {
// haltedTraversers in execute() represent worker-traversal traversers
// for example, from a traversal of the form g.V().out().program(...)
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
index 6bd2a04..012b9fc 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
@@ -58,9 +58,12 @@ import org.apache.tinkerpop.gremlin.process.computer.util.DefaultComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
import org.apache.tinkerpop.gremlin.structure.io.Storage;
import org.apache.tinkerpop.gremlin.util.Gremlin;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.io.File;
import java.io.NotSerializableException;
+import java.util.HashSet;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
@@ -74,6 +77,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
protected GiraphConfiguration giraphConfiguration = new GiraphConfiguration();
private MapMemory memory = new MapMemory();
private boolean useWorkerThreadsInConfiguration;
+ private Set<String> vertexProgramConfigurationKeys = new HashSet<>();
public GiraphGraphComputer(final HadoopGraph hadoopGraph) {
super(hadoopGraph);
@@ -112,6 +116,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
final BaseConfiguration apacheConfiguration = new BaseConfiguration();
apacheConfiguration.setDelimiterParsingDisabled(true);
vertexProgram.storeState(apacheConfiguration);
+ IteratorUtils.fill(apacheConfiguration.getKeys(), this.vertexProgramConfigurationKeys);
ConfUtil.mergeApacheIntoHadoopConfiguration(apacheConfiguration, this.giraphConfiguration);
this.vertexProgram.getMessageCombiner().ifPresent(combiner -> this.giraphConfiguration.setMessageCombinerClass(GiraphMessageCombiner.class));
return this;
@@ -139,6 +144,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
// clear properties that should not be propagated in an OLAP chain
apacheConfiguration.clearProperty(Constants.GREMLIN_HADOOP_GRAPH_FILTER);
apacheConfiguration.clearProperty(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR);
+ this.vertexProgramConfigurationKeys.forEach(apacheConfiguration::clearProperty); // clear out vertex program specific configurations
return new DefaultComputerResult(InputOutputHelper.getOutputGraph(apacheConfiguration, this.resultGraph, this.persist), this.memory.asImmutable());
}, exec);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
index 1327a7f..23d33f1 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
@@ -192,7 +192,6 @@ public final class MemoryTraversalSideEffects implements TraversalSideEffects {
traversal.getSideEffects();
sideEffects.keys().
stream().
- filter(key -> !key.equals(TraversalVertexProgram.HALTED_TRAVERSERS)).
forEach(key -> keys.add(MemoryComputeKey.of(key, sideEffects.getReducer(key), true, false)));
return keys;
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
index 4df5189..211a5e5 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
@@ -33,6 +33,7 @@ import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.Computer
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.VertexProgramStrategy;
import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
+import org.apache.tinkerpop.gremlin.process.computer.util.VertexProgramHelper;
import org.apache.tinkerpop.gremlin.process.traversal.Operator;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
@@ -66,6 +67,7 @@ import org.apache.tinkerpop.gremlin.util.iterator.EmptyIterator;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
@@ -102,6 +104,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
private PureTraversal<?, ?> traversal;
private TraversalMatrix<?, ?> traversalMatrix;
private final Set<MapReduce> mapReducers = new HashSet<>();
+ private TraverserSet<Object> haltedTraversers;
private boolean returnHaltedTraversers = false;
private TraversalVertexProgram() {
@@ -116,6 +119,32 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
return this.traversal;
}
+ public static <R> TraverserSet<R> loadHaltedTraversers(final Configuration configuration) {
+ if (!configuration.containsKey(HALTED_TRAVERSERS))
+ return new TraverserSet<>();
+
+ final Object object = configuration.getProperty(HALTED_TRAVERSERS) instanceof String ?
+ VertexProgramHelper.deserialize(configuration, HALTED_TRAVERSERS) :
+ configuration.getProperty(HALTED_TRAVERSERS);
+ if (object instanceof Traverser.Admin)
+ return new TraverserSet<>((Traverser.Admin<R>) object);
+ else {
+ final TraverserSet<R> traverserSet = new TraverserSet<>();
+ traverserSet.addAll((Collection) object);
+ return traverserSet;
+ }
+ }
+
+ public static <R> void storeHaltedTraversers(final Configuration configuration, final TraverserSet<R> haltedTraversers) {
+ if (null != haltedTraversers && !haltedTraversers.isEmpty()) {
+ try {
+ VertexProgramHelper.serialize(haltedTraversers, configuration, HALTED_TRAVERSERS);
+ } catch (final Exception e) {
+ configuration.setProperty(HALTED_TRAVERSERS, haltedTraversers);
+ }
+ }
+ }
+
@Override
public void loadState(final Graph graph, final Configuration configuration) {
if (!configuration.containsKey(TRAVERSAL))
@@ -125,6 +154,8 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
this.traversal.get().applyStrategies();
/// traversal is compiled and ready to be introspected
this.traversalMatrix = new TraversalMatrix<>(this.traversal.get());
+ // get any master-traversal halted traversers
+ this.haltedTraversers = TraversalVertexProgram.loadHaltedTraversers(configuration);
// if results will be serialized out, don't save halted traversers across the cluster
this.returnHaltedTraversers =
(this.traversal.get().getParent().asStep().getNextStep() instanceof ComputerResultStep || // if its just going to stream it out, don't distribute
@@ -158,6 +189,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
public void storeState(final Configuration configuration) {
VertexProgram.super.storeState(configuration);
this.traversal.storeState(configuration, TRAVERSAL);
+ TraversalVertexProgram.storeHaltedTraversers(configuration, this.haltedTraversers);
}
@Override
@@ -170,19 +202,17 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
memory.set(MUTATED_MEMORY_KEYS, new HashSet<>());
memory.set(COMPLETED_BARRIERS, new HashSet<>());
// if halted traversers are being sent from a previous VertexProgram in an OLAP chain (non-distributed traversers), get them into the stream
- if (sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS)) {
- final TraverserSet<Object> haltedTraversers = sideEffects.get(TraversalVertexProgram.HALTED_TRAVERSERS);
+ if (null != this.haltedTraversers) {
final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
- IteratorUtils.removeOnNext(haltedTraversers.iterator()).forEachRemaining(traverser -> {
+ IteratorUtils.removeOnNext(this.haltedTraversers.iterator()).forEachRemaining(traverser -> {
traverser.setStepId(this.traversal.get().getStartStep().getId());
toProcessTraversers.add(traverser);
});
assert haltedTraversers.isEmpty(); // TODO: this should be empty
final TraverserSet<Object> remoteActiveTraversers = new TraverserSet<>();
- MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, haltedTraversers);
- memory.set(HALTED_TRAVERSERS, haltedTraversers);
+ MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, this.haltedTraversers);
+ memory.set(HALTED_TRAVERSERS, this.haltedTraversers);
memory.set(ACTIVE_TRAVERSERS, remoteActiveTraversers);
- sideEffects.remove(TraversalVertexProgram.HALTED_TRAVERSERS);
} else {
memory.set(HALTED_TRAVERSERS, new TraverserSet<>());
memory.set(ACTIVE_TRAVERSERS, new TraverserSet<>());
@@ -196,6 +226,9 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
@Override
public void execute(final Vertex vertex, final Messenger<TraverserSet<Object>> messenger, final Memory memory) {
+ // if any global halted traversers, simply don't use them as they were handled by master setup()
+ if (null != this.haltedTraversers)
+ this.haltedTraversers = null;
// memory is distributed
MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.EXECUTE);
// if a barrier was completed in another worker, it is also completed here (ensure distributed barriers are synchronized)
@@ -365,6 +398,11 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
super(TraversalVertexProgram.class);
}
+ public Builder haltedTraversers(final TraverserSet<Object> haltedTraversers) {
+ TraversalVertexProgram.storeHaltedTraversers(this.configuration, haltedTraversers);
+ return this;
+ }
+
public Builder traversal(final TraversalSource traversalSource, final String scriptEngine, final String traversalScript, final Object... bindings) {
return this.traversal(new ScriptTraversal<>(traversalSource, scriptEngine, traversalScript, bindings));
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
index 167924f..f5facdb 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
@@ -87,9 +87,16 @@ public final class TraverserExecutor {
while (traversers.hasNext()) {
final Traverser.Admin<Object> traverser = traversers.next();
traversers.remove();
- traverser.attach(Attachable.Method.get(vertex));
- traverser.setSideEffects(traversalSideEffects);
- toProcessTraversers.add(traverser);
+ if (traverser.isHalted()) {
+ if (returnHaltedTraversers)
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
+ else
+ haltedTraversers.add(traverser);
+ } else {
+ traverser.attach(Attachable.Method.get(vertex));
+ traverser.setSideEffects(traversalSideEffects);
+ toProcessTraversers.add(traverser);
+ }
}
}
@@ -152,10 +159,10 @@ public final class TraverserExecutor {
final TraverserSet<Object> barrierSet = barrier.nextBarrier();
IteratorUtils.removeOnNext(barrierSet.iterator()).forEachRemaining(traverser -> {
traverser.addLabels(step.getLabels()); // this might need to be generalized for working with global barriers too
- if (traverser.isHalted()) {
+ if (traverser.isHalted() && ((!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) || getHostingVertex(traverser.get()).equals(vertex))) {
traverser.detach();
if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser.split()));
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
else
haltedTraversers.add(traverser);
} else {
@@ -174,10 +181,10 @@ public final class TraverserExecutor {
}
} else { // LOCAL PROCESSING
step.forEachRemaining(traverser -> {
- if (traverser.isHalted()) {
+ if (traverser.isHalted() && ((!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) || getHostingVertex(traverser.get()).equals(vertex))) {
traverser.detach();
if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser.split()));
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
else
haltedTraversers.add(traverser);
} else {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
index 57e1a3e..b599c42 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
@@ -21,9 +21,12 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal.step;
import org.apache.tinkerpop.gremlin.process.computer.Computer;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.structure.Graph;
+import java.util.Optional;
+
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
@@ -48,10 +51,11 @@ public interface VertexComputing {
/**
* Generate the {@link VertexProgram}.
*
- * @param graph the {@link Graph} that the program will be executed over.
+ * @param graph the {@link Graph} that the program will be executed over.
+ * @param memoryOptional a reference to a {@link Memory} from the previous OLAP job if it exists.
* @return the generated vertex program instance.
*/
- public VertexProgram generateProgram(final Graph graph);
+ public VertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional);
/**
* @deprecated As of release 3.2.1. Please use {@link VertexComputing#getComputer()}.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
index 5d10e67..660f5b8 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
@@ -19,8 +19,9 @@
package org.apache.tinkerpop.gremlin.process.computer.traversal.step.map;
-import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.lambda.HaltedTraversersCountTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies;
@@ -37,8 +38,8 @@ import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
-import java.util.function.Function;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -83,7 +84,7 @@ public final class PageRankVertexProgramStep extends VertexProgramStep implement
}
@Override
- public PageRankVertexProgram generateProgram(final Graph graph) {
+ public PageRankVertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional) {
final Traversal.Admin<Vertex, Edge> detachedTraversal = this.edgeTraversal.getPure();
detachedTraversal.setStrategies(TraversalStrategies.GlobalCache.getStrategies(graph.getClass()));
final PageRankVertexProgram.Builder builder = PageRankVertexProgram.build()
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
index f42ce93..2ae36a5 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
@@ -19,7 +19,7 @@
package org.apache.tinkerpop.gremlin.process.computer.traversal.step.map;
-import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.clustering.peerpressure.PeerPressureVertexProgram;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies;
@@ -36,6 +36,7 @@ import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
/**
@@ -84,9 +85,15 @@ public final class PeerPressureVertexProgramStep extends VertexProgramStep imple
}
@Override
- public PeerPressureVertexProgram generateProgram(final Graph graph) {
+ public PeerPressureVertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional) {
final Traversal.Admin<Vertex, Edge> detachedTraversal = this.edgeTraversal.getPure();
detachedTraversal.setStrategies(TraversalStrategies.GlobalCache.getStrategies(graph.getClass()));
+ /*
+ memoryOptional.ifPresent(memory -> {
+ if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
+ builder.configure(TraversalVertexProgram.HALTED_TRAVERSERS, memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
+ });
+ */
return PeerPressureVertexProgram.build()
.property(this.clusterProperty)
.maxIterations(this.times)
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
index a1342b6..d060c99 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
@@ -20,7 +20,9 @@
package org.apache.tinkerpop.gremlin.process.computer.traversal.step.map;
import org.apache.commons.configuration.MapConfiguration;
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.PureTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
@@ -29,6 +31,7 @@ import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -48,11 +51,16 @@ public final class ProgramVertexProgramStep extends VertexProgramStep {
}
@Override
- public VertexProgram generateProgram(final Graph graph) {
+ public VertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional) {
final MapConfiguration base = new MapConfiguration(this.configuration);
base.setDelimiterParsingDisabled(true);
PureTraversal.storeState(base, ROOT_TRAVERSAL, TraversalHelper.getRootTraversal(this.getTraversal()).clone());
base.setProperty(STEP_ID, this.getId());
+ memoryOptional.ifPresent(memory -> {
+ if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS)) {
+ TraversalVertexProgram.storeHaltedTraversers(base, memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
+ }
+ });
return VertexProgram.createVertexProgram(graph, base);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
index 64b9097..5851f35 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
@@ -20,6 +20,7 @@
package org.apache.tinkerpop.gremlin.process.computer.traversal.step.map;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.traversal.MemoryTraversalSideEffects;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
@@ -32,6 +33,7 @@ import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
/**
@@ -62,15 +64,18 @@ public final class TraversalVertexProgramStep extends VertexProgramStep implemen
}
@Override
- public TraversalVertexProgram generateProgram(final Graph graph) {
+ public TraversalVertexProgram generateProgram(final Graph graph, final Optional<Memory> memoryOptional) {
final Traversal.Admin<?, ?> computerSpecificTraversal = this.computerTraversal.getPure();
computerSpecificTraversal.setStrategies(TraversalStrategies.GlobalCache.getStrategies(graph.getClass()).clone());
this.getTraversal().getStrategies().toList().forEach(computerSpecificTraversal.getStrategies()::addStrategies);
computerSpecificTraversal.setSideEffects(new MemoryTraversalSideEffects(this.getTraversal().getSideEffects()));
computerSpecificTraversal.setParent(this);
- return TraversalVertexProgram.build()
- .traversal(computerSpecificTraversal)
- .create(graph);
+ final TraversalVertexProgram.Builder builder = TraversalVertexProgram.build().traversal(computerSpecificTraversal);
+ memoryOptional.ifPresent(memory -> {
+ if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
+ builder.haltedTraversers(memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
+ });
+ return builder.create(graph);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
index c535d30..ac3de30 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
@@ -25,7 +25,6 @@ import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.VertexComputing;
-import org.apache.tinkerpop.gremlin.process.traversal.Operator;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
@@ -33,12 +32,11 @@ import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.AbstractStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.ProfileStep;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.util.function.ConstantSupplier;
import java.util.NoSuchElementException;
+import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -66,14 +64,15 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com
if (this.first && this.getPreviousStep() instanceof EmptyStep) {
this.first = false;
final Graph graph = this.getTraversal().getGraph().get();
- future = this.generateComputer(graph).program(this.generateProgram(graph)).submit();
+ future = this.generateComputer(graph).program(this.generateProgram(graph, Optional.empty())).submit();
final ComputerResult result = future.get();
this.processMemorySideEffects(result.memory());
return this.getTraversal().getTraverserGenerator().generate(result, this, 1l);
} else {
final Traverser.Admin<ComputerResult> traverser = this.starts.next();
final Graph graph = traverser.get().graph();
- future = this.generateComputer(graph).program(this.generateProgram(graph)).submit();
+ final Memory memory = traverser.get().memory();
+ future = this.generateComputer(graph).program(this.generateProgram(graph, Optional.of(memory))).submit();
final ComputerResult result = future.get();
this.processMemorySideEffects(result.memory());
return traverser.split(result, this);
@@ -122,21 +121,13 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com
final TraversalSideEffects sideEffects = this.getTraversal().getSideEffects();
for (final String key : memory.keys()) {
if (sideEffects.exists(key)) {
+ // halted traversers should never be propagated through sideEffects
+ assert !key.equals(TraversalVertexProgram.HALTED_TRAVERSERS);
sideEffects.set(key, memory.get(key));
}
}
- if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS) && !this.isEndStep()) {
- final TraverserSet<Object> haltedTraversers = memory.get(TraversalVertexProgram.HALTED_TRAVERSERS);
- if (!haltedTraversers.isEmpty()) {
- if (sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
- sideEffects.set(TraversalVertexProgram.HALTED_TRAVERSERS, haltedTraversers);
- else
- sideEffects.register(TraversalVertexProgram.HALTED_TRAVERSERS, new ConstantSupplier<>(haltedTraversers), Operator.addAll);
- }
- }
}
-
protected boolean isEndStep() {
return this.getNextStep() instanceof ComputerResultStep || (this.getNextStep() instanceof ProfileStep && this.getNextStep().getNextStep() instanceof ComputerResultStep);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java
index 90819d7..e4cf5ff 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/optimization/GraphFilterStrategy.java
@@ -41,6 +41,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
/**
@@ -59,7 +60,7 @@ public final class GraphFilterStrategy extends AbstractTraversalStrategy<Travers
return;
final Graph graph = traversal.getGraph().orElse(EmptyGraph.instance()); // given that this strategy only works for single OLAP jobs, the graph is the traversal graph
for (final TraversalVertexProgramStep step : TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal)) { // will be zero or one step
- final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph).getTraversal().get().clone();
+ final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, Optional.empty()).getTraversal().get().clone();
if (!computerTraversal.isLocked())
computerTraversal.applyStrategies();
final Computer computer = step.getComputer();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ImmutablePath.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ImmutablePath.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ImmutablePath.java
index 1a916ca..1e936c8 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ImmutablePath.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ImmutablePath.java
@@ -20,7 +20,6 @@ package org.apache.tinkerpop.gremlin.process.traversal.step.util;
import org.apache.tinkerpop.gremlin.process.traversal.Path;
import org.apache.tinkerpop.gremlin.process.traversal.Pop;
-import org.apache.tinkerpop.gremlin.structure.Element;
import java.io.Serializable;
import java.util.ArrayList;
@@ -177,7 +176,7 @@ public class ImmutablePath implements Path, ImmutablePathImpl, Serializable, Clo
}
- private static class TailPath implements Path, ImmutablePathImpl {
+ private static class TailPath implements Path, ImmutablePathImpl, Serializable {
private static final TailPath INSTANCE = new TailPath();
private TailPath() {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
----------------------------------------------------------------------
diff --git a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
index bda6819..e27e8c0 100644
--- a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
+++ b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPageRankTest.groovy
@@ -69,5 +69,10 @@ public abstract class GroovyPageRankTest {
public Traversal<Vertex, Map<Object, List<Vertex>>> get_g_V_outXcreatedX_groupXmX_byXlabelX_pageRankX1X_byXpageRankX_byXinEX_timesX1X_inXcreatedX_groupXmX_byXpageRankX_capXmX() {
new ScriptTraversal<>(g, "gremlin-groovy", "g.V.out('created').group('m').by(label).pageRank(1.0).by('pageRank').by(inE()).times(1).in('created').group('m').by('pageRank').cap('m')")
}
+
+ @Override
+ public Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX() {
+ new ScriptTraversal<>(g, "gremlin-groovy", "g.V.out('created').pageRank().by(bothE()).by('projectRank').valueMap('name','projectRank')")
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
index df4b802..5a3ed9b 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PageRankTest.java
@@ -31,6 +31,7 @@ import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.junit.Test;
import org.junit.runner.RunWith;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -48,6 +49,8 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
public abstract Traversal<Vertex, Vertex> get_g_V_pageRank();
+ public abstract Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX();
+
public abstract Traversal<Vertex, String> get_g_V_pageRank_order_byXpageRank_decrX_name();
public abstract Traversal<Vertex, String> get_g_V_pageRank_order_byXpageRank_decrX_name_limitX2X();
@@ -78,6 +81,20 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
@Test
@LoadGraphWith(MODERN)
+ public void g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX() {
+ final Traversal<Vertex, Map<String, Object>> traversal = get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX();
+ printTraversalForm(traversal);
+ final Map<String, Double> map = new HashMap<>();
+ traversal.forEachRemaining(m -> map.put(((List<String>) m.get("name")).get(0), ((List<Double>) m.get("projectRank")).get(0)));
+ assertEquals(2, map.size());
+ assertTrue(map.containsKey("lop"));
+ assertTrue(map.containsKey("ripple"));
+ assertTrue(map.get("lop") > map.get("ripple"));
+ assertFalse(traversal.hasNext());
+ }
+
+ @Test
+ @LoadGraphWith(MODERN)
public void g_V_pageRank_order_byXpageRank_decrX_name() {
final Traversal<Vertex, String> traversal = get_g_V_pageRank_order_byXpageRank_decrX_name();
printTraversalForm(traversal);
@@ -225,6 +242,11 @@ public abstract class PageRankTest extends AbstractGremlinProcessTest {
}
@Override
+ public Traversal<Vertex, Map<String, Object>> get_g_V_outXcreatedX_pageRank_byXbothEX_byXprojectRankX_valueMapXname_projectRankX() {
+ return g.V().out("created").pageRank().by(__.bothE()).by("projectRank").valueMap("name", "projectRank");
+ }
+
+ @Override
public Traversal<Vertex, Map<String, List<Object>>> get_g_V_pageRank_byXoutEXknowsXX_byXfriendRankX_valueMapXname_friendRankX() {
return g.V().pageRank().by(__.outE("knows")).by("friendRank").valueMap("name", "friendRank");
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
index 2c505e7..8b2a293 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
@@ -59,6 +59,7 @@ import java.util.Set;
import static org.apache.tinkerpop.gremlin.LoadGraphWith.GraphData.MODERN;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
/**
@@ -160,6 +161,7 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
public static class TestProgram implements VertexProgram {
private PureTraversal<?, ?> traversal = new PureTraversal<>(EmptyTraversal.instance());
+ private TraverserSet<Object> haltedTraversers;
private Step programStep = EmptyStep.instance();
private final Set<MemoryComputeKey> memoryComputeKeys = new HashSet<>();
@@ -168,6 +170,7 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
public void loadState(final Graph graph, final Configuration configuration) {
VertexProgram.super.loadState(graph, configuration);
this.traversal = PureTraversal.loadState(configuration, VertexProgramStep.ROOT_TRAVERSAL, graph);
+ this.haltedTraversers = TraversalVertexProgram.loadHaltedTraversers(configuration);
this.programStep = new TraversalMatrix<>(this.traversal.get()).getStepById(configuration.getString(ProgramVertexProgramStep.STEP_ID));
this.memoryComputeKeys.addAll(MemoryTraversalSideEffects.getMemoryComputeKeys(this.traversal.get()));
this.memoryComputeKeys.add(MemoryComputeKey.of(TraversalVertexProgram.HALTED_TRAVERSERS, Operator.addAll, false, false));
@@ -178,24 +181,24 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
public void storeState(final Configuration configuration) {
VertexProgram.super.storeState(configuration);
this.traversal.storeState(configuration, VertexProgramStep.ROOT_TRAVERSAL);
+ TraversalVertexProgram.storeHaltedTraversers(configuration, this.haltedTraversers);
configuration.setProperty(ProgramVertexProgramStep.STEP_ID, this.programStep.getId());
}
@Override
public void setup(final Memory memory) {
MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.SETUP);
- final TraversalSideEffects sideEffects = this.traversal.get().getSideEffects();
- final TraverserSet<Object> haltedTraversers = sideEffects.get(TraversalVertexProgram.HALTED_TRAVERSERS);
- sideEffects.remove(TraversalVertexProgram.HALTED_TRAVERSERS);
- this.checkSideEffects(MemoryTraversalSideEffects.State.SETUP);
- final Map<Vertex, Long> map = (Map<Vertex, Long>) haltedTraversers.iterator().next().get();
+ final Map<Vertex, Long> map = (Map<Vertex, Long>) this.haltedTraversers.iterator().next().get();
assertEquals(2, map.size());
assertTrue(map.values().contains(3l));
assertTrue(map.values().contains(1l));
final TraverserSet<Object> activeTraversers = new TraverserSet<>();
- map.keySet().forEach(vertex -> activeTraversers.add(haltedTraversers.peek().split(vertex, EmptyStep.instance())));
+ map.keySet().forEach(vertex -> activeTraversers.add(this.haltedTraversers.peek().split(vertex, EmptyStep.instance())));
+ this.haltedTraversers.clear();
+ this.checkSideEffects();
memory.set(TraversalVertexProgram.ACTIVE_TRAVERSERS, activeTraversers);
+
}
@Override
@@ -203,7 +206,7 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
final TraverserGenerator generator = this.traversal.get().getTraverserGenerator();
MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.EXECUTE);
- this.checkSideEffects(MemoryTraversalSideEffects.State.EXECUTE);
+ this.checkSideEffects();
final TraverserSet<Vertex> activeTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
if (vertex.label().equals("software")) {
assertEquals(1, activeTraversers.stream().filter(v -> v.get().equals(vertex)).count());
@@ -231,7 +234,7 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
public boolean terminate(final Memory memory) {
final TraverserGenerator generator = this.traversal.get().getTraverserGenerator();
MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.TERMINATE);
- checkSideEffects(MemoryTraversalSideEffects.State.TERMINATE);
+ checkSideEffects();
if (memory.isInitialIteration()) {
assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
return false;
@@ -248,16 +251,18 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
@Override
public void workerIterationStart(final Memory memory) {
+ assertNotNull(this.haltedTraversers);
+ this.haltedTraversers.clear();
assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.WORKER_ITERATION_START);
- checkSideEffects(MemoryTraversalSideEffects.State.WORKER_ITERATION_START);
+ checkSideEffects();
}
@Override
public void workerIterationEnd(final Memory memory) {
assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.WORKER_ITERATION_END);
- checkSideEffects(MemoryTraversalSideEffects.State.WORKER_ITERATION_END);
+ checkSideEffects();
}
@Override
@@ -299,16 +304,13 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
////////
- private void checkSideEffects(final MemoryTraversalSideEffects.State state) {
+ private void checkSideEffects() {
+ assertEquals(0, this.haltedTraversers.size());
+ assertTrue(this.haltedTraversers.isEmpty());
final TraversalSideEffects sideEffects = this.traversal.get().getSideEffects();
assertTrue(sideEffects instanceof MemoryTraversalSideEffects);
- if (state.masterState()) {
- assertEquals(1, sideEffects.keys().size());
- assertFalse(sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
- } else {
- assertEquals(2, sideEffects.keys().size());
- assertTrue(sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
- }
+ assertEquals(1, sideEffects.keys().size());
+ assertFalse(sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
assertTrue(sideEffects.exists("x"));
final BulkSet<String> bulkSet = sideEffects.get("x");
assertEquals(4, bulkSet.size());
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/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 9a9e934..9e05e53 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
@@ -264,14 +264,13 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
throw new IllegalStateException(e.getMessage());
}
} else { // standard GraphComputer semantics
+ // get a configuration that will be propagated to all workers
+ final HadoopConfiguration vertexProgramConfiguration = new HadoopConfiguration();
+ this.vertexProgram.storeState(vertexProgramConfiguration);
// 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();
- this.vertexProgram.storeState(vertexProgramConfiguration);
- ConfigurationUtils.copy(vertexProgramConfiguration, apacheConfiguration);
- ConfUtil.mergeApacheIntoHadoopConfiguration(vertexProgramConfiguration, hadoopConfiguration);
// execute the vertex program
while (true) {
if (Thread.interrupted()) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/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
index 19d21b3..a2cc518 100644
--- 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
@@ -20,6 +20,7 @@
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.Memory;
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;
@@ -29,6 +30,8 @@ import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.op
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
+import java.util.Optional;
+
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
@@ -43,7 +46,7 @@ public final class SparkInterceptorStrategy extends AbstractTraversalStrategy<Tr
public void apply(final Traversal.Admin<?, ?> traversal) {
final Graph graph = traversal.getGraph().orElse(EmptyGraph.instance()); // best guess at what the graph will be as its dynamically determined
for (final TraversalVertexProgramStep step : TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal)) {
- final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph).getTraversal().get().clone();
+ final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, Optional.empty()).getTraversal().get().clone();
if (!computerTraversal.isLocked())
computerTraversal.applyStrategies();
if (SparkStarBarrierInterceptor.isLegal(computerTraversal)) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4ee30289/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java
index 153282e..280c7ba 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/SparkSingleIterationStrategy.java
@@ -20,6 +20,7 @@
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.Memory;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
import org.apache.tinkerpop.gremlin.process.traversal.Scope;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
@@ -36,6 +37,7 @@ import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
import java.util.Arrays;
import java.util.HashSet;
+import java.util.Optional;
import java.util.Set;
/**
@@ -59,7 +61,7 @@ public final class SparkSingleIterationStrategy extends AbstractTraversalStrateg
public void apply(final Traversal.Admin<?, ?> traversal) {
final Graph graph = traversal.getGraph().orElse(EmptyGraph.instance()); // best guess at what the graph will be as its dynamically determined
for (final TraversalVertexProgramStep step : TraversalHelper.getStepsOfClass(TraversalVertexProgramStep.class, traversal)) {
- final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph).getTraversal().get().clone();
+ final Traversal.Admin<?, ?> computerTraversal = step.generateProgram(graph, Optional.empty()).getTraversal().get().clone();
if (!computerTraversal.isLocked())
computerTraversal.applyStrategies();
boolean doesMessagePass = TraversalHelper.hasStepOfAssignableClassRecursively(Scope.global, MULTI_ITERATION_CLASSES, computerTraversal);
[05/20] incubator-tinkerpop git commit: minor tweaks to
TraverserVertexProgram to make it more efficient.
Posted by dk...@apache.org.
minor tweaks to TraverserVertexProgram to make it more efficient.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/a6aacdc5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/a6aacdc5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/a6aacdc5
Branch: refs/heads/TINKERPOP-1298
Commit: a6aacdc51eba895e04f9d75d64343470fbaca5c5
Parents: 07e1299
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 18 17:11:26 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 18 17:11:26 2016 -0600
----------------------------------------------------------------------
.../process/computer/traversal/TraversalVertexProgram.java | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a6aacdc5/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
index 211a5e5..aba44d3 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
@@ -196,19 +196,18 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
public void setup(final Memory memory) {
// memory is local
MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.SETUP);
- final MemoryTraversalSideEffects sideEffects = ((MemoryTraversalSideEffects) this.traversal.get().getSideEffects());
- sideEffects.storeSideEffectsInMemory();
+ ((MemoryTraversalSideEffects) this.traversal.get().getSideEffects()).storeSideEffectsInMemory();
memory.set(VOTE_TO_HALT, true);
memory.set(MUTATED_MEMORY_KEYS, new HashSet<>());
memory.set(COMPLETED_BARRIERS, new HashSet<>());
// if halted traversers are being sent from a previous VertexProgram in an OLAP chain (non-distributed traversers), get them into the stream
- if (null != this.haltedTraversers) {
+ if (null != this.haltedTraversers && !this.haltedTraversers.isEmpty()) {
final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
IteratorUtils.removeOnNext(this.haltedTraversers.iterator()).forEachRemaining(traverser -> {
traverser.setStepId(this.traversal.get().getStartStep().getId());
toProcessTraversers.add(traverser);
});
- assert haltedTraversers.isEmpty(); // TODO: this should be empty
+ assert haltedTraversers.isEmpty();
final TraverserSet<Object> remoteActiveTraversers = new TraverserSet<>();
MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, this.haltedTraversers);
memory.set(HALTED_TRAVERSERS, this.haltedTraversers);
@@ -217,6 +216,8 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
memory.set(HALTED_TRAVERSERS, new TraverserSet<>());
memory.set(ACTIVE_TRAVERSERS, new TraverserSet<>());
}
+ // local variable will no longer be used so null it for GC
+ this.haltedTraversers = null;
}
@Override
[09/20] incubator-tinkerpop git commit: added traversal-based vote
strength test to PeerPressureTest. Cleaned up internal classes of
TraversalVertexProgram a bit for better organization. Generalized Phase enum
so it can be used by other computer-based cl
Posted by dk...@apache.org.
added traversal-based vote strength test to PeerPressureTest. Cleaned up internal classes of TraversalVertexProgram a bit for better organization. Generalized Phase enum so it can be used by other computer-based classes. added comments to TraveralVertexProgram.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/5bcf0b01
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/5bcf0b01
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/5bcf0b01
Branch: refs/heads/TINKERPOP-1298
Commit: 5bcf0b01c7bfc77e9eea8fdb7b2c4c54a4fc0000
Parents: b100f03
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu May 19 05:23:36 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu May 19 05:23:36 2016 -0600
----------------------------------------------------------------------
docs/src/reference/the-traversal.asciidoc | 17 +-
.../gremlin/process/computer/ProgramPhase.java | 40 ++++
.../peerpressure/PeerPressureVertexProgram.java | 21 +-
.../traversal/MemoryTraversalSideEffects.java | 27 +--
.../computer/traversal/SingleMessenger.java | 49 -----
.../traversal/TraversalVertexProgram.java | 18 +-
.../computer/traversal/TraverserExecutor.java | 216 ------------------
.../computer/traversal/WorkerExecutor.java | 220 +++++++++++++++++++
.../step/map/PeerPressureVertexProgramStep.java | 9 +-
.../process/computer/util/EmptyMemory.java | 58 ++++-
.../process/computer/util/SingleMessenger.java | 49 +++++
.../step/map/GroovyPeerPressureTest.groovy | 5 +
.../traversal/step/map/PeerPressureTest.java | 31 +++
.../process/traversal/step/map/ProgramTest.java | 11 +-
.../SparkStarBarrierInterceptor.java | 3 +-
15 files changed, 460 insertions(+), 314 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/docs/src/reference/the-traversal.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/reference/the-traversal.asciidoc b/docs/src/reference/the-traversal.asciidoc
index 3394e90..4c74795 100644
--- a/docs/src/reference/the-traversal.asciidoc
+++ b/docs/src/reference/the-traversal.asciidoc
@@ -1434,6 +1434,10 @@ vertex program includes:
The user supplied `VertexProgram` can leverage that information accordingly within their vertex program. Example uses
are provided below.
+WARNING: Developing a `VertexProgram` is for expert users. Moreover, developing one that can be used effectively within
+a traversal requires yet more expertise. This information is recommended to advanced users with a deep understanding of the
+mechanics of Gremlin OLAP (<<graphcomputer,`GraphComputer`>>).
+
[source,java]
----
private TraverserSet<Object> haltedTraversers;
@@ -1447,7 +1451,7 @@ public void loadState(final Graph graph, final Configuration configuration) {
// if master-traversal traversers may be propagated, create a memory compute key
this.memoryComputeKeys.add(MemoryComputeKey.of(TraversalVertexProgram.HALTED_TRAVERSERS, Operator.addAll, false, false));
// returns an empty traverser set if there are no halted traversers
- this.haltedTraversers = TraversalVertexProgram.getHaltedTraversers(configuration);
+ this.haltedTraversers = TraversalVertexProgram.loadHaltedTraversers(configuration);
}
public void storeState(final Configuration configuration) {
@@ -1457,22 +1461,21 @@ public void storeState(final Configuration configuration) {
}
public void setup(final Memory memory) {
- if(null != this.haltedTraversers) {
+ if(!this.haltedTraversers.isEmpty()) {
// do what you like with the halted master traversal traversers
}
// once used, no need to keep that information around (master)
- if(null != this.haltedTraversers)
- this.haltedTraversers.clear()
+ this.haltedTraversers = null;
}
public void execute(final Vertex vertex, final Messenger messenger, final Memory memory) {
// once used, no need to keep that information around (workers)
if(null != this.haltedTraversers)
- this.haltedTraversers.clear()
+ this.haltedTraversers = null;
if(vertex.property(TraversalVertexProgram.HALTED_TRAVERSERS).isPresent()) {
// haltedTraversers in execute() represent worker-traversal traversers
// for example, from a traversal of the form g.V().out().program(...)
- TraverserSet<Object> haltedTraversers = vertex.value(TraversalVertexProgram.HALTED_TRAVERSERS) :
+ TraverserSet<Object> haltedTraversers = vertex.value(TraversalVertexProgram.HALTED_TRAVERSERS);
// create a new halted traverser set that can be used by the next OLAP job in the chain
// these are worker-traversers that are distributed throughout the graph
TraverserSet<Object> newHaltedTraversers = new TraverserSet<>();
@@ -1480,7 +1483,7 @@ public void execute(final Vertex vertex, final Messenger messenger, final Memory
newHaltedTraversers.add(traverser.split(traverser.get().toString(), this.programStep));
});
vertex.property(VertexProperty.Cardinality.single, TraversalVertexProgram.HALTED_TRAVERSERS, newHaltedTraversers);
- // it is possible to create master-traversers that are localized to the master traversal (thread)
+ // it is possible to create master-traversers that are localized to the master traversal (this is how results are ultimately delivered back to the user)
memory.add(TraversalVertexProgram.HALTED_TRAVERSERS,
new TraverserSet<>(this.traversal().get().getTraverserGenerator().generate("an example", this.programStep, 1l)));
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ProgramPhase.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ProgramPhase.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ProgramPhase.java
new file mode 100644
index 0000000..ce39505
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ProgramPhase.java
@@ -0,0 +1,40 @@
+/*
+ * 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;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public enum ProgramPhase {
+
+ SETUP,
+ WORKER_ITERATION_START,
+ EXECUTE,
+ WORKER_ITERATION_END,
+ TERMINATE;
+
+ public boolean masterState() {
+ return this == SETUP || this == TERMINATE;
+ }
+
+ public boolean workerState() {
+ return !this.masterState();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
index 8834882..56de255 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
@@ -34,6 +34,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.MapHelper;
import org.apache.tinkerpop.gremlin.process.traversal.util.PureTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.ScriptTraversal;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
@@ -62,6 +63,7 @@ public class PeerPressureVertexProgram extends StaticVertexProgram<Pair<Serializ
public static final String CLUSTER = "gremlin.peerPressureVertexProgram.cluster";
private static final String VOTE_STRENGTH = "gremlin.peerPressureVertexProgram.voteStrength";
+ private static final String INITIAL_VOTE_STRENGTH_TRAVERSAL = "gremlin.pageRankVertexProgram.initialVoteStrengthTraversal";
private static final String PROPERTY = "gremlin.peerPressureVertexProgram.property";
private static final String MAX_ITERATIONS = "gremlin.peerPressureVertexProgram.maxIterations";
private static final String DISTRIBUTE_VOTE = "gremlin.peerPressureVertexProgram.distributeVote";
@@ -69,6 +71,7 @@ public class PeerPressureVertexProgram extends StaticVertexProgram<Pair<Serializ
private static final String VOTE_TO_HALT = "gremlin.peerPressureVertexProgram.voteToHalt";
private PureTraversal<Vertex, Edge> edgeTraversal = null;
+ private PureTraversal<Vertex, ? extends Number> initialVoteStrengthTraversal = null;
private int maxIterations = 30;
private boolean distributeVote = false;
private String property = CLUSTER;
@@ -81,6 +84,8 @@ public class PeerPressureVertexProgram extends StaticVertexProgram<Pair<Serializ
@Override
public void loadState(final Graph graph, final Configuration configuration) {
+ if (configuration.containsKey(INITIAL_VOTE_STRENGTH_TRAVERSAL))
+ this.initialVoteStrengthTraversal = PureTraversal.loadState(configuration, INITIAL_VOTE_STRENGTH_TRAVERSAL, graph);
if (configuration.containsKey(EDGE_TRAVERSAL)) {
this.edgeTraversal = PureTraversal.loadState(configuration, EDGE_TRAVERSAL, graph);
this.voteScope = MessageScope.Local.of(() -> this.edgeTraversal.get().clone());
@@ -99,6 +104,8 @@ public class PeerPressureVertexProgram extends StaticVertexProgram<Pair<Serializ
configuration.setProperty(DISTRIBUTE_VOTE, this.distributeVote);
if (null != this.edgeTraversal)
this.edgeTraversal.storeState(configuration, EDGE_TRAVERSAL);
+ if (null != this.initialVoteStrengthTraversal)
+ this.initialVoteStrengthTraversal.storeState(configuration, INITIAL_VOTE_STRENGTH_TRAVERSAL);
}
@Override
@@ -137,14 +144,19 @@ public class PeerPressureVertexProgram extends StaticVertexProgram<Pair<Serializ
if (this.distributeVote) {
messenger.sendMessage(this.countScope, Pair.with('c', 1.0d));
} else {
- double voteStrength = 1.0d;
+ double voteStrength = (null == this.initialVoteStrengthTraversal ?
+ 1.0d :
+ TraversalUtil.apply(vertex, this.initialVoteStrengthTraversal.get()).doubleValue());
vertex.property(VertexProperty.Cardinality.single, this.property, vertex.id());
vertex.property(VertexProperty.Cardinality.single, VOTE_STRENGTH, voteStrength);
messenger.sendMessage(this.voteScope, new Pair<>((Serializable) vertex.id(), voteStrength));
memory.add(VOTE_TO_HALT, false);
}
} else if (1 == memory.getIteration() && this.distributeVote) {
- double voteStrength = 1.0d / IteratorUtils.reduce(IteratorUtils.map(messenger.receiveMessages(), Pair::getValue1), 0.0d, (a, b) -> a + b);
+ double voteStrength = (null == this.initialVoteStrengthTraversal ?
+ 1.0d :
+ TraversalUtil.apply(vertex, this.initialVoteStrengthTraversal.get()).doubleValue()) /
+ IteratorUtils.reduce(IteratorUtils.map(messenger.receiveMessages(), Pair::getValue1), 0.0d, (a, b) -> a + b);
vertex.property(VertexProperty.Cardinality.single, this.property, vertex.id());
vertex.property(VertexProperty.Cardinality.single, VOTE_STRENGTH, voteStrength);
messenger.sendMessage(this.voteScope, new Pair<>((Serializable) vertex.id(), voteStrength));
@@ -227,6 +239,11 @@ public class PeerPressureVertexProgram extends StaticVertexProgram<Pair<Serializ
return this;
}
+ public Builder initialVoteStrength(final Traversal.Admin<Vertex, ? extends Number> initialVoteStrengthTraversal) {
+ PureTraversal.storeState(this.configuration, INITIAL_VOTE_STRENGTH_TRAVERSAL, initialVoteStrengthTraversal);
+ return this;
+ }
+
/**
* @deprecated As of release 3.2.0, replaced by {@link PeerPressureVertexProgram.Builder#edges(Traversal.Admin)}
*/
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
index 23d33f1..bf9f8c0 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MemoryComputeKey;
+import org.apache.tinkerpop.gremlin.process.computer.ProgramPhase;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
@@ -38,23 +39,7 @@ public final class MemoryTraversalSideEffects implements TraversalSideEffects {
private TraversalSideEffects sideEffects;
private Memory memory;
- private State state;
-
- public enum State {
- SETUP,
- WORKER_ITERATION_START,
- EXECUTE,
- WORKER_ITERATION_END,
- TERMINATE;
-
- public boolean masterState() {
- return this == SETUP || this == TERMINATE;
- }
-
- public boolean workerState() {
- return !this.masterState();
- }
- }
+ private ProgramPhase phase;
private MemoryTraversalSideEffects() {
// for serialization
@@ -93,7 +78,7 @@ public final class MemoryTraversalSideEffects implements TraversalSideEffects {
@Override
public void add(final String key, final Object value) {
- if (this.state.workerState())
+ if (this.phase.workerState())
this.memory.add(key, value);
else
this.memory.set(key, this.sideEffects.getReducer(key).apply(this.memory.get(key), value));
@@ -168,20 +153,20 @@ public final class MemoryTraversalSideEffects implements TraversalSideEffects {
}
public void storeSideEffectsInMemory() {
- if (this.state.workerState())
+ if (this.phase.workerState())
this.sideEffects.forEach(this.memory::add);
else
this.sideEffects.forEach(this.memory::set);
}
- public static void setMemorySideEffects(final Traversal.Admin<?, ?> traversal, final Memory memory, final State state) {
+ public static void setMemorySideEffects(final Traversal.Admin<?, ?> traversal, final Memory memory, final ProgramPhase phase) {
final TraversalSideEffects sideEffects = traversal.getSideEffects();
if (!(sideEffects instanceof MemoryTraversalSideEffects)) {
traversal.setSideEffects(new MemoryTraversalSideEffects(sideEffects));
}
final MemoryTraversalSideEffects memoryTraversalSideEffects = ((MemoryTraversalSideEffects) traversal.getSideEffects());
memoryTraversalSideEffects.memory = memory;
- memoryTraversalSideEffects.state = state;
+ memoryTraversalSideEffects.phase = phase;
}
public static Set<MemoryComputeKey> getMemoryComputeKeys(final Traversal.Admin<?, ?> traversal) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/SingleMessenger.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/SingleMessenger.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/SingleMessenger.java
deleted file mode 100644
index 26ed8a4..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/SingleMessenger.java
+++ /dev/null
@@ -1,49 +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.traversal;
-
-import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
-import org.apache.tinkerpop.gremlin.process.computer.Messenger;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-
-import java.util.Iterator;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class SingleMessenger<M> implements Messenger<M> {
-
- private final Messenger<M> baseMessenger;
- private final M message;
-
- public SingleMessenger(final Messenger<M> baseMessenger, final M message) {
- this.baseMessenger = baseMessenger;
- this.message = message;
- }
-
- @Override
- public Iterator<M> receiveMessages() {
- return IteratorUtils.of(this.message);
- }
-
- @Override
- public void sendMessage(final MessageScope messageScope, final M message) {
- this.baseMessenger.sendMessage(messageScope, message);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
index c7e7ef9..266426f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
@@ -27,12 +27,14 @@ import org.apache.tinkerpop.gremlin.process.computer.MemoryComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.MessageCombiner;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
+import org.apache.tinkerpop.gremlin.process.computer.ProgramPhase;
import org.apache.tinkerpop.gremlin.process.computer.VertexComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.ComputerResultStep;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.VertexProgramStrategy;
import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
+import org.apache.tinkerpop.gremlin.process.computer.util.SingleMessenger;
import org.apache.tinkerpop.gremlin.process.computer.util.VertexProgramHelper;
import org.apache.tinkerpop.gremlin.process.traversal.Operator;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
@@ -195,13 +197,13 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
@Override
public void setup(final Memory memory) {
// memory is local
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.SETUP);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.SETUP);
((MemoryTraversalSideEffects) this.traversal.get().getSideEffects()).storeSideEffectsInMemory();
memory.set(VOTE_TO_HALT, true);
memory.set(MUTATED_MEMORY_KEYS, new HashSet<>());
memory.set(COMPLETED_BARRIERS, new HashSet<>());
- // if halted traversers are being sent from a previous VertexProgram in an OLAP chain (non-distributed traversers), get them into the stream
- if (null != this.haltedTraversers && !this.haltedTraversers.isEmpty()) {
+ // if halted traversers are being sent from a previous VertexProgram in an OLAP chain (non-distributed traversers), get them into the flow
+ if (!this.haltedTraversers.isEmpty()) {
final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
IteratorUtils.removeOnNext(this.haltedTraversers.iterator()).forEachRemaining(traverser -> {
traverser.setStepId(this.traversal.get().getStartStep().getId());
@@ -231,7 +233,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
if (null != this.haltedTraversers)
this.haltedTraversers = null;
// memory is distributed
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.EXECUTE);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.EXECUTE);
// if a barrier was completed in another worker, it is also completed here (ensure distributed barriers are synchronized)
final Set<String> completedBarriers = memory.get(COMPLETED_BARRIERS);
for (final String stepId : completedBarriers) {
@@ -245,11 +247,13 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
//////////////////
if (memory.isInitialIteration()) { // ITERATION 1
final TraverserSet<Object> activeTraversers = new TraverserSet<>();
+ // if halted traversers are being sent from a previous VertexProgram in an OLAP chain (distributed traversers), get them into the flow
IteratorUtils.removeOnNext(haltedTraversers.iterator()).forEachRemaining(traverser -> {
traverser.setStepId(this.traversal.get().getStartStep().getId());
activeTraversers.add(traverser);
});
assert haltedTraversers.isEmpty();
+ // for g.V()/E()
if (this.traversal.get().getStartStep() instanceof GraphStep) {
final GraphStep<Element, Element> graphStep = (GraphStep<Element, Element>) this.traversal.get().getStartStep();
graphStep.reset();
@@ -270,9 +274,9 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
activeTraversers.add((Traverser.Admin) traverser);
});
}
- memory.add(VOTE_TO_HALT, activeTraversers.isEmpty() || TraverserExecutor.execute(vertex, new SingleMessenger<>(messenger, activeTraversers), this.traversalMatrix, memory, this.returnHaltedTraversers));
+ memory.add(VOTE_TO_HALT, activeTraversers.isEmpty() || WorkerExecutor.execute(vertex, new SingleMessenger<>(messenger, activeTraversers), this.traversalMatrix, memory, this.returnHaltedTraversers));
} else { // ITERATION 1+
- memory.add(VOTE_TO_HALT, TraverserExecutor.execute(vertex, messenger, this.traversalMatrix, memory, this.returnHaltedTraversers));
+ memory.add(VOTE_TO_HALT, WorkerExecutor.execute(vertex, messenger, this.traversalMatrix, memory, this.returnHaltedTraversers));
}
if (this.returnHaltedTraversers || haltedTraversers.isEmpty())
vertex.<TraverserSet>property(HALTED_TRAVERSERS).remove();
@@ -281,7 +285,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
@Override
public boolean terminate(final Memory memory) {
// memory is local
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.TERMINATE);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.TERMINATE);
final boolean voteToHalt = memory.<Boolean>get(VOTE_TO_HALT);
memory.set(VOTE_TO_HALT, true);
memory.set(ACTIVE_TRAVERSERS, new TraverserSet<>());
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
deleted file mode 100644
index c5f9bb5..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
+++ /dev/null
@@ -1,216 +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.traversal;
-
-import org.apache.tinkerpop.gremlin.process.computer.Memory;
-import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
-import org.apache.tinkerpop.gremlin.process.computer.Messenger;
-import org.apache.tinkerpop.gremlin.process.traversal.Step;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
-import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Bypassing;
-import org.apache.tinkerpop.gremlin.process.traversal.step.LocalBarrier;
-import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
-import org.apache.tinkerpop.gremlin.structure.Edge;
-import org.apache.tinkerpop.gremlin.structure.Element;
-import org.apache.tinkerpop.gremlin.structure.Property;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.util.Attachable;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class TraverserExecutor {
-
- public static boolean execute(final Vertex vertex, final Messenger<TraverserSet<Object>> messenger, final TraversalMatrix<?, ?> traversalMatrix, final Memory memory, final boolean returnHaltedTraversers) {
-
- final TraversalSideEffects traversalSideEffects = traversalMatrix.getTraversal().getSideEffects();
- final AtomicBoolean voteToHalt = new AtomicBoolean(true);
- final TraverserSet<Object> haltedTraversers = vertex.value(TraversalVertexProgram.HALTED_TRAVERSERS);
- final TraverserSet<Object> activeTraversers = new TraverserSet<>();
- final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
-
- ////////////////////////////////
- // GENERATE LOCAL TRAVERSERS //
- ///////////////////////////////
-
- // these are traversers that are going from OLTP to OLAP
- final TraverserSet<Object> maybeActiveTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
- final Iterator<Traverser.Admin<Object>> iterator = maybeActiveTraversers.iterator();
- while (iterator.hasNext()) {
- final Traverser.Admin<Object> traverser = iterator.next();
- if (vertex.equals(TraverserExecutor.getHostingVertex(traverser.get()))) {
- // iterator.remove(); ConcurrentModificationException
- traverser.attach(Attachable.Method.get(vertex));
- traverser.setSideEffects(traversalSideEffects);
- toProcessTraversers.add(traverser);
- }
- }
- // these are traversers that exist from from a local barrier
- vertex.<TraverserSet<Object>>property(TraversalVertexProgram.ACTIVE_TRAVERSERS).ifPresent(previousActiveTraversers -> {
- IteratorUtils.removeOnNext(previousActiveTraversers.iterator()).forEachRemaining(traverser -> {
- traverser.attach(Attachable.Method.get(vertex));
- traverser.setSideEffects(traversalSideEffects);
- toProcessTraversers.add(traverser);
- });
- vertex.property(TraversalVertexProgram.ACTIVE_TRAVERSERS).remove();
- });
- // these are traversers that have been messaged to the vertex
- final Iterator<TraverserSet<Object>> messages = messenger.receiveMessages();
- while (messages.hasNext()) {
- final Iterator<Traverser.Admin<Object>> traversers = messages.next().iterator();
- while (traversers.hasNext()) {
- final Traverser.Admin<Object> traverser = traversers.next();
- traversers.remove();
- if (traverser.isHalted()) {
- if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
- else
- haltedTraversers.add(traverser);
- } else {
- traverser.attach(Attachable.Method.get(vertex));
- traverser.setSideEffects(traversalSideEffects);
- toProcessTraversers.add(traverser);
- }
- }
- }
-
- ///////////////////////////////
- // PROCESS LOCAL TRAVERSERS //
- //////////////////////////////
-
- // while there are still local traversers, process them until they leave the vertex or halt (i.e. isHalted()).
- while (!toProcessTraversers.isEmpty()) {
- // process local traversers and if alive, repeat, else halt.
- Step<Object, Object> previousStep = EmptyStep.instance();
- Iterator<Traverser.Admin<Object>> traversers = toProcessTraversers.iterator();
- while (traversers.hasNext()) {
- final Traverser.Admin<Object> traverser = traversers.next();
- traversers.remove();
- final Step<Object, Object> currentStep = traversalMatrix.getStepById(traverser.getStepId());
- if (!currentStep.getId().equals(previousStep.getId()) && !(previousStep instanceof EmptyStep))
- TraverserExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers);
- currentStep.addStart(traverser);
- previousStep = currentStep;
- }
- TraverserExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers);
- assert toProcessTraversers.isEmpty();
- // process all the local objects and send messages or store locally again
- if (!activeTraversers.isEmpty()) {
- traversers = activeTraversers.iterator();
- while (traversers.hasNext()) {
- final Traverser.Admin<Object> traverser = traversers.next();
- traversers.remove();
- if (traverser.get() instanceof Element || traverser.get() instanceof Property) { // GRAPH OBJECT
- // if the element is remote, then message, else store it locally for re-processing
- final Vertex hostingVertex = TraverserExecutor.getHostingVertex(traverser.get());
- if (!vertex.equals(hostingVertex)) { // necessary for path access
- voteToHalt.set(false);
- traverser.detach();
- messenger.sendMessage(MessageScope.Global.of(hostingVertex), new TraverserSet<>(traverser));
- } else {
- if (traverser.get() instanceof Attachable) // necessary for path access to local object
- traverser.attach(Attachable.Method.get(vertex));
- toProcessTraversers.add(traverser);
- }
- } else // STANDARD OBJECT
- toProcessTraversers.add(traverser);
- }
- assert activeTraversers.isEmpty();
- }
- }
- return voteToHalt.get();
- }
-
- private static void drainStep(final Vertex vertex, final Step<Object, Object> step, final TraverserSet<Object> activeTraversers, final TraverserSet<Object> haltedTraversers, final Memory memory, final boolean returnHaltedTraversers) {
- if (step instanceof Barrier) {
- if (step instanceof Bypassing)
- ((Bypassing) step).setBypass(true);
- if (step instanceof LocalBarrier) {
- final LocalBarrier<Object> barrier = (LocalBarrier<Object>) step;
- final TraverserSet<Object> traverserSet = vertex.<TraverserSet<Object>>property(TraversalVertexProgram.ACTIVE_TRAVERSERS).orElse(new TraverserSet<>());
- vertex.property(TraversalVertexProgram.ACTIVE_TRAVERSERS, traverserSet);
- while (barrier.hasNextBarrier()) {
- final TraverserSet<Object> barrierSet = barrier.nextBarrier();
- IteratorUtils.removeOnNext(barrierSet.iterator()).forEachRemaining(traverser -> {
- traverser.addLabels(step.getLabels()); // this might need to be generalized for working with global barriers too
- if (traverser.isHalted() &&
- (returnHaltedTraversers ||
- (!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
- getHostingVertex(traverser.get()).equals(vertex))) {
- traverser.detach();
- if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
- else
- haltedTraversers.add(traverser);
- } else {
- traverser.detach();
- traverserSet.add(traverser);
- }
- });
- }
- memory.add(TraversalVertexProgram.MUTATED_MEMORY_KEYS, new HashSet<>(Collections.singleton(step.getId())));
- } else {
- final Barrier barrier = (Barrier) step;
- while (barrier.hasNextBarrier()) {
- memory.add(step.getId(), barrier.nextBarrier());
- }
- memory.add(TraversalVertexProgram.MUTATED_MEMORY_KEYS, new HashSet<>(Collections.singleton(step.getId())));
- }
- } else { // LOCAL PROCESSING
- step.forEachRemaining(traverser -> {
- if (traverser.isHalted() &&
- (returnHaltedTraversers ||
- (!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
- getHostingVertex(traverser.get()).equals(vertex))) {
- traverser.detach();
- if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
- else
- haltedTraversers.add(traverser);
- } else {
- activeTraversers.add(traverser);
- }
- });
- }
- }
-
- private static Vertex getHostingVertex(final Object object) {
- Object obj = object;
- while (true) {
- if (obj instanceof Vertex)
- return (Vertex) obj;
- else if (obj instanceof Edge)
- return ((Edge) obj).outVertex();
- else if (obj instanceof Property)
- obj = ((Property) obj).element();
- else
- throw new IllegalStateException("The host of the object is unknown: " + obj.toString() + ':' + obj.getClass().getCanonicalName());
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
new file mode 100644
index 0000000..c4bd659
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
@@ -0,0 +1,220 @@
+/*
+ * 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;
+
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
+import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
+import org.apache.tinkerpop.gremlin.process.computer.Messenger;
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Bypassing;
+import org.apache.tinkerpop.gremlin.process.traversal.step.LocalBarrier;
+import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
+import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Element;
+import org.apache.tinkerpop.gremlin.structure.Property;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.util.Attachable;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+final class WorkerExecutor {
+
+ private WorkerExecutor() {
+
+ }
+
+ protected static boolean execute(final Vertex vertex, final Messenger<TraverserSet<Object>> messenger, final TraversalMatrix<?, ?> traversalMatrix, final Memory memory, final boolean returnHaltedTraversers) {
+
+ final TraversalSideEffects traversalSideEffects = traversalMatrix.getTraversal().getSideEffects();
+ final AtomicBoolean voteToHalt = new AtomicBoolean(true);
+ final TraverserSet<Object> haltedTraversers = vertex.value(TraversalVertexProgram.HALTED_TRAVERSERS);
+ final TraverserSet<Object> activeTraversers = new TraverserSet<>();
+ final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
+
+ ////////////////////////////////
+ // GENERATE LOCAL TRAVERSERS //
+ ///////////////////////////////
+
+ // these are traversers that are going from OLTP to OLAP
+ final TraverserSet<Object> maybeActiveTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
+ final Iterator<Traverser.Admin<Object>> iterator = maybeActiveTraversers.iterator();
+ while (iterator.hasNext()) {
+ final Traverser.Admin<Object> traverser = iterator.next();
+ if (vertex.equals(WorkerExecutor.getHostingVertex(traverser.get()))) {
+ // iterator.remove(); ConcurrentModificationException
+ traverser.attach(Attachable.Method.get(vertex));
+ traverser.setSideEffects(traversalSideEffects);
+ toProcessTraversers.add(traverser);
+ }
+ }
+ // these are traversers that exist from from a local barrier
+ vertex.<TraverserSet<Object>>property(TraversalVertexProgram.ACTIVE_TRAVERSERS).ifPresent(previousActiveTraversers -> {
+ IteratorUtils.removeOnNext(previousActiveTraversers.iterator()).forEachRemaining(traverser -> {
+ traverser.attach(Attachable.Method.get(vertex));
+ traverser.setSideEffects(traversalSideEffects);
+ toProcessTraversers.add(traverser);
+ });
+ vertex.property(TraversalVertexProgram.ACTIVE_TRAVERSERS).remove();
+ });
+ // these are traversers that have been messaged to the vertex
+ final Iterator<TraverserSet<Object>> messages = messenger.receiveMessages();
+ while (messages.hasNext()) {
+ final Iterator<Traverser.Admin<Object>> traversers = messages.next().iterator();
+ while (traversers.hasNext()) {
+ final Traverser.Admin<Object> traverser = traversers.next();
+ traversers.remove();
+ if (traverser.isHalted()) {
+ if (returnHaltedTraversers)
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
+ else
+ haltedTraversers.add(traverser);
+ } else {
+ traverser.attach(Attachable.Method.get(vertex));
+ traverser.setSideEffects(traversalSideEffects);
+ toProcessTraversers.add(traverser);
+ }
+ }
+ }
+
+ ///////////////////////////////
+ // PROCESS LOCAL TRAVERSERS //
+ //////////////////////////////
+
+ // while there are still local traversers, process them until they leave the vertex or halt (i.e. isHalted()).
+ while (!toProcessTraversers.isEmpty()) {
+ // process local traversers and if alive, repeat, else halt.
+ Step<Object, Object> previousStep = EmptyStep.instance();
+ Iterator<Traverser.Admin<Object>> traversers = toProcessTraversers.iterator();
+ while (traversers.hasNext()) {
+ final Traverser.Admin<Object> traverser = traversers.next();
+ traversers.remove();
+ final Step<Object, Object> currentStep = traversalMatrix.getStepById(traverser.getStepId());
+ if (!currentStep.getId().equals(previousStep.getId()) && !(previousStep instanceof EmptyStep))
+ WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers);
+ currentStep.addStart(traverser);
+ previousStep = currentStep;
+ }
+ WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers);
+ assert toProcessTraversers.isEmpty();
+ // process all the local objects and send messages or store locally again
+ if (!activeTraversers.isEmpty()) {
+ traversers = activeTraversers.iterator();
+ while (traversers.hasNext()) {
+ final Traverser.Admin<Object> traverser = traversers.next();
+ traversers.remove();
+ if (traverser.get() instanceof Element || traverser.get() instanceof Property) { // GRAPH OBJECT
+ // if the element is remote, then message, else store it locally for re-processing
+ final Vertex hostingVertex = WorkerExecutor.getHostingVertex(traverser.get());
+ if (!vertex.equals(hostingVertex)) { // necessary for path access
+ voteToHalt.set(false);
+ traverser.detach();
+ messenger.sendMessage(MessageScope.Global.of(hostingVertex), new TraverserSet<>(traverser));
+ } else {
+ if (traverser.get() instanceof Attachable) // necessary for path access to local object
+ traverser.attach(Attachable.Method.get(vertex));
+ toProcessTraversers.add(traverser);
+ }
+ } else // STANDARD OBJECT
+ toProcessTraversers.add(traverser);
+ }
+ assert activeTraversers.isEmpty();
+ }
+ }
+ return voteToHalt.get();
+ }
+
+ private static void drainStep(final Vertex vertex, final Step<Object, Object> step, final TraverserSet<Object> activeTraversers, final TraverserSet<Object> haltedTraversers, final Memory memory, final boolean returnHaltedTraversers) {
+ if (step instanceof Barrier) {
+ if (step instanceof Bypassing)
+ ((Bypassing) step).setBypass(true);
+ if (step instanceof LocalBarrier) {
+ final LocalBarrier<Object> barrier = (LocalBarrier<Object>) step;
+ final TraverserSet<Object> traverserSet = vertex.<TraverserSet<Object>>property(TraversalVertexProgram.ACTIVE_TRAVERSERS).orElse(new TraverserSet<>());
+ vertex.property(TraversalVertexProgram.ACTIVE_TRAVERSERS, traverserSet);
+ while (barrier.hasNextBarrier()) {
+ final TraverserSet<Object> barrierSet = barrier.nextBarrier();
+ IteratorUtils.removeOnNext(barrierSet.iterator()).forEachRemaining(traverser -> {
+ traverser.addLabels(step.getLabels()); // this might need to be generalized for working with global barriers too
+ if (traverser.isHalted() &&
+ (returnHaltedTraversers ||
+ (!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
+ getHostingVertex(traverser.get()).equals(vertex))) {
+ traverser.detach();
+ if (returnHaltedTraversers)
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
+ else
+ haltedTraversers.add(traverser);
+ } else {
+ traverser.detach();
+ traverserSet.add(traverser);
+ }
+ });
+ }
+ memory.add(TraversalVertexProgram.MUTATED_MEMORY_KEYS, new HashSet<>(Collections.singleton(step.getId())));
+ } else {
+ final Barrier barrier = (Barrier) step;
+ while (barrier.hasNextBarrier()) {
+ memory.add(step.getId(), barrier.nextBarrier());
+ }
+ memory.add(TraversalVertexProgram.MUTATED_MEMORY_KEYS, new HashSet<>(Collections.singleton(step.getId())));
+ }
+ } else { // LOCAL PROCESSING
+ step.forEachRemaining(traverser -> {
+ if (traverser.isHalted() &&
+ (returnHaltedTraversers ||
+ (!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
+ getHostingVertex(traverser.get()).equals(vertex))) {
+ traverser.detach();
+ if (returnHaltedTraversers)
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
+ else
+ haltedTraversers.add(traverser);
+ } else {
+ activeTraversers.add(traverser);
+ }
+ });
+ }
+ }
+
+ private static Vertex getHostingVertex(final Object object) {
+ Object obj = object;
+ while (true) {
+ if (obj instanceof Vertex)
+ return (Vertex) obj;
+ else if (obj instanceof Edge)
+ return ((Edge) obj).outVertex();
+ else if (obj instanceof Property)
+ obj = ((Property) obj).element();
+ else
+ throw new IllegalStateException("The host of the object is unknown: " + obj.toString() + ':' + obj.getClass().getCanonicalName());
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
index 7bd726e..0ea5112 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal.step.map;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.clustering.peerpressure.PeerPressureVertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.lambda.HaltedTraversersCountTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
@@ -87,11 +88,13 @@ public final class PeerPressureVertexProgramStep extends VertexProgramStep imple
public PeerPressureVertexProgram generateProgram(final Graph graph, final Memory memory) {
final Traversal.Admin<Vertex, Edge> detachedTraversal = this.edgeTraversal.getPure();
detachedTraversal.setStrategies(TraversalStrategies.GlobalCache.getStrategies(graph.getClass()));
- return PeerPressureVertexProgram.build()
+ final PeerPressureVertexProgram.Builder builder = PeerPressureVertexProgram.build()
.property(this.clusterProperty)
.maxIterations(this.times)
- .edges(detachedTraversal)
- .create(graph);
+ .edges(detachedTraversal);
+ if (this.previousTraversalVertexProgram())
+ builder.initialVoteStrength(new HaltedTraversersCountTraversal());
+ return builder.create(graph);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/EmptyMemory.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/EmptyMemory.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/EmptyMemory.java
index 72b1bbf..f513ee6 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/EmptyMemory.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/EmptyMemory.java
@@ -21,19 +21,71 @@ package org.apache.tinkerpop.gremlin.process.computer.util;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
+import java.util.Collections;
+import java.util.Set;
+
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public final class EmptyMemory {
+public final class EmptyMemory implements Memory.Admin {
- private static final Memory INSTANCE = new ImmutableMemory(new MapMemory());
+ private static final EmptyMemory INSTANCE = new EmptyMemory();
private EmptyMemory() {
}
- public static Memory instance() {
+ public static EmptyMemory instance() {
return INSTANCE;
}
+ @Override
+ public void setIteration(final int iteration) {
+ throw Memory.Exceptions.memoryIsCurrentlyImmutable();
+ }
+
+ @Override
+ public void setRuntime(final long runtime) {
+ throw Memory.Exceptions.memoryIsCurrentlyImmutable();
+ }
+
+ @Override
+ public Memory asImmutable() {
+ return this;
+ }
+
+ @Override
+ public Set<String> keys() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public <R> R get(final String key) throws IllegalArgumentException {
+ throw Memory.Exceptions.memoryDoesNotExist(key);
+ }
+
+ @Override
+ public void set(final String key, final Object value) throws IllegalArgumentException, IllegalStateException {
+ throw Memory.Exceptions.memoryIsCurrentlyImmutable();
+ }
+
+ @Override
+ public void add(final String key, final Object value) throws IllegalArgumentException, IllegalStateException {
+ throw Memory.Exceptions.memoryIsCurrentlyImmutable();
+ }
+
+ @Override
+ public int getIteration() {
+ return 0;
+ }
+
+ @Override
+ public long getRuntime() {
+ return 0;
+ }
+
+ @Override
+ public boolean exists(final String key) {
+ return false;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/SingleMessenger.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/SingleMessenger.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/SingleMessenger.java
new file mode 100644
index 0000000..63ee282
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/SingleMessenger.java
@@ -0,0 +1,49 @@
+/*
+ * 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.MessageScope;
+import org.apache.tinkerpop.gremlin.process.computer.Messenger;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
+
+import java.util.Iterator;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class SingleMessenger<M> implements Messenger<M> {
+
+ private final Messenger<M> baseMessenger;
+ private final M message;
+
+ public SingleMessenger(final Messenger<M> baseMessenger, final M message) {
+ this.baseMessenger = baseMessenger;
+ this.message = message;
+ }
+
+ @Override
+ public Iterator<M> receiveMessages() {
+ return IteratorUtils.of(this.message);
+ }
+
+ @Override
+ public void sendMessage(final MessageScope messageScope, final M message) {
+ this.baseMessenger.sendMessage(messageScope, message);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPeerPressureTest.groovy
----------------------------------------------------------------------
diff --git a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPeerPressureTest.groovy b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPeerPressureTest.groovy
index eb1ad67..6ec0750 100644
--- a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPeerPressureTest.groovy
+++ b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroovyPeerPressureTest.groovy
@@ -39,5 +39,10 @@ public abstract class GroovyPeerPressureTest {
public Traversal<Vertex, Map<Object, Number>> get_g_V_peerPressure_byXclusterX_byXoutEXknowsXX_pageRankX1X_byXrankX_byXoutEXknowsXX_timesX2X_group_byXclusterX_byXrank_sumX_limitX100X() {
new ScriptTraversal<>(g, "gremlin-groovy", "g.V.peerPressure.by('cluster').by(outE('knows')).pageRank(1.0).by('rank').by(outE('knows')).times(1).group.by('cluster').by(values('rank').sum).limit(100)")
}
+
+ @Override
+ public Traversal<Vertex, Map<String, List<Object>>> get_g_V_hasXname_rippleX_inXcreatedX_peerPressure_byXoutEX_byXclusterX_repeatXunionXidentity__bothX_timesX2X_dedup_valueMapXname_clusterX() {
+ new ScriptTraversal<>(g, "gremlin-groovy", "g.V.has('name', 'ripple').in('created').peerPressure.by(outE()).by('cluster').repeat(union(identity(), both())).times(2).dedup.valueMap('name', 'cluster')")
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PeerPressureTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PeerPressureTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PeerPressureTest.java
index 21c4f43..5a2477c 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PeerPressureTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/PeerPressureTest.java
@@ -27,7 +27,11 @@ import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.junit.Test;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
import static org.apache.tinkerpop.gremlin.LoadGraphWith.GraphData.MODERN;
import static org.junit.Assert.assertEquals;
@@ -43,6 +47,7 @@ public abstract class PeerPressureTest extends AbstractGremlinProcessTest {
public abstract Traversal<Vertex, Map<Object, Number>> get_g_V_peerPressure_byXclusterX_byXoutEXknowsXX_pageRankX1X_byXrankX_byXoutEXknowsXX_timesX2X_group_byXclusterX_byXrank_sumX_limitX100X();
+ public abstract Traversal<Vertex, Map<String, List<Object>>> get_g_V_hasXname_rippleX_inXcreatedX_peerPressure_byXoutEX_byXclusterX_repeatXunionXidentity__bothX_timesX2X_dedup_valueMapXname_clusterX();
@Test
@LoadGraphWith(MODERN)
@@ -72,6 +77,27 @@ public abstract class PeerPressureTest extends AbstractGremlinProcessTest {
assertEquals(0.0d, (double) map.get(convertToVertexId("peter")), 0.001d);
}
+ @Test
+ @LoadGraphWith(MODERN)
+ public void g_V_hasXname_rippleX_inXcreatedX_peerPressure_byXoutEX_byXclusterX_repeatXunionXidentity__bothX_timesX2X_dedup_valueMapXname_clusterX() {
+ final Traversal<Vertex, Map<String, List<Object>>> traversal = get_g_V_hasXname_rippleX_inXcreatedX_peerPressure_byXoutEX_byXclusterX_repeatXunionXidentity__bothX_timesX2X_dedup_valueMapXname_clusterX();
+ printTraversalForm(traversal);
+ final List<Map<String, List<Object>>> results = traversal.toList();
+ assertEquals(6, results.size());
+ final Map<String, Object> clusters = new HashMap<>();
+ results.forEach(m -> clusters.put((String) m.get("name").get(0), m.get("cluster").get(0)));
+ assertEquals(2, results.get(0).size());
+ assertEquals(6, clusters.size());
+ assertEquals(clusters.get("josh"), clusters.get("ripple"));
+ assertEquals(clusters.get("josh"), clusters.get("lop"));
+ final Set<Object> ids = new HashSet<>(clusters.values());
+ assertEquals(4, ids.size());
+ assertTrue(ids.contains(convertToVertexId("marko")));
+ assertTrue(ids.contains(convertToVertexId("vadas")));
+ assertTrue(ids.contains(convertToVertexId("josh")));
+ assertTrue(ids.contains(convertToVertexId("peter")));
+ }
+
public static class Traversals extends PeerPressureTest {
@@ -84,5 +110,10 @@ public abstract class PeerPressureTest extends AbstractGremlinProcessTest {
public Traversal<Vertex, Map<Object, Number>> get_g_V_peerPressure_byXclusterX_byXoutEXknowsXX_pageRankX1X_byXrankX_byXoutEXknowsXX_timesX2X_group_byXclusterX_byXrank_sumX_limitX100X() {
return g.V().peerPressure().by("cluster").by(__.outE("knows")).pageRank(1.0d).by("rank").by(__.outE("knows")).times(1).<Object, Number>group().by("cluster").by(__.values("rank").sum()).limit(100);
}
+
+ @Override
+ public Traversal<Vertex, Map<String, List<Object>>> get_g_V_hasXname_rippleX_inXcreatedX_peerPressure_byXoutEX_byXclusterX_repeatXunionXidentity__bothX_timesX2X_dedup_valueMapXname_clusterX() {
+ return g.V().has("name", "ripple").in("created").peerPressure().by(__.outE()).by("cluster").repeat(__.union(__.identity(), __.both())).times(2).dedup().valueMap("name", "cluster");
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
index 8b2a293..5f548da 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
@@ -27,6 +27,7 @@ import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MemoryComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
+import org.apache.tinkerpop.gremlin.process.computer.ProgramPhase;
import org.apache.tinkerpop.gremlin.process.computer.VertexComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgram;
@@ -187,7 +188,7 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
@Override
public void setup(final Memory memory) {
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.SETUP);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.SETUP);
final Map<Vertex, Long> map = (Map<Vertex, Long>) this.haltedTraversers.iterator().next().get();
assertEquals(2, map.size());
assertTrue(map.values().contains(3l));
@@ -205,7 +206,7 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
public void execute(final Vertex vertex, final Messenger messenger, final Memory memory) {
assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
final TraverserGenerator generator = this.traversal.get().getTraverserGenerator();
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.EXECUTE);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.EXECUTE);
this.checkSideEffects();
final TraverserSet<Vertex> activeTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
if (vertex.label().equals("software")) {
@@ -233,7 +234,7 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
@Override
public boolean terminate(final Memory memory) {
final TraverserGenerator generator = this.traversal.get().getTraverserGenerator();
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.TERMINATE);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.TERMINATE);
checkSideEffects();
if (memory.isInitialIteration()) {
assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
@@ -254,14 +255,14 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
assertNotNull(this.haltedTraversers);
this.haltedTraversers.clear();
assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.WORKER_ITERATION_START);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.WORKER_ITERATION_START);
checkSideEffects();
}
@Override
public void workerIterationEnd(final Memory memory) {
assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.WORKER_ITERATION_END);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.WORKER_ITERATION_END);
checkSideEffects();
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5bcf0b01/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java
index 6e35cf8..768d10a 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java
@@ -22,6 +22,7 @@ package org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.o
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.process.computer.ProgramPhase;
import org.apache.tinkerpop.gremlin.process.computer.traversal.MemoryTraversalSideEffects;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.traversal.NumberHelper;
@@ -75,7 +76,7 @@ public final class SparkStarBarrierInterceptor implements SparkVertexProgramInte
traversal.applyStrategies(); // compile
boolean identityTraversal = traversal.getSteps().isEmpty(); // if the traversal is empty, just return the vertex (fast)
///////////////////////////////
- MemoryTraversalSideEffects.setMemorySideEffects(traversal, memory, MemoryTraversalSideEffects.State.EXECUTE); // any intermediate sideEffect steps are backed by SparkMemory
+ MemoryTraversalSideEffects.setMemorySideEffects(traversal, memory, ProgramPhase.EXECUTE); // any intermediate sideEffect steps are backed by SparkMemory
memory.setInExecute(true);
final JavaRDD<Traverser.Admin<Object>> nextRDD = inputRDD.values()
.filter(vertexWritable -> ElementHelper.idExists(vertexWritable.get().id(), graphStepIds)) // ensure vertex ids are in V(x)