You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ok...@apache.org on 2016/02/06 20:54:38 UTC
[07/30] incubator-tinkerpop git commit: GraphFilter is now a really
cool class. It is part of gremlin-core/computer and provides access to
GraphComputer vertices() and edges() load filters. It also provides direct
support for filtering StarVertex vertice
GraphFilter is now a really cool class. It is part of gremlin-core/computer and provides access to GraphComputer vertices() and edges() load filters. It also provides direct support for filtering StarVertex vertices (as most OLAP systems will leverage StarVertex). Its StarVertex support is nice in that GraphFilter analyzes the edgesFilter and can do bulk dropEdges() to prune the StarVertex fast. Whatever it can't do in bulk, it then runs the edgeFilter over the remaining edges. GraphComputerTest.shouldSupportGraphFilter() ensures that the graph is properly pruned. I have some ideas about pushing GraphFilter down to the StarVertex deserializer, but will need @spmallette help on that. If we can do that, then we can get some BLAZING speeds for highly pruned OLAP operations.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/d0ac6527
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/d0ac6527
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/d0ac6527
Branch: refs/heads/master
Commit: d0ac65277702b703c1ab2257adcbf67b0699b959
Parents: bc417db
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue Feb 2 08:55:14 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue Feb 2 08:55:14 2016 -0700
----------------------------------------------------------------------
.../structure/io/GiraphVertexInputFormat.java | 2 +-
.../giraph/structure/io/GiraphVertexReader.java | 3 +-
.../gremlin/process/computer/GraphComputer.java | 2 +-
.../gremlin/process/computer/GraphFilter.java | 94 +++++++++++++++++---
.../gremlin/structure/util/star/StarGraph.java | 60 +++++++++++--
.../process/computer/GraphComputerTest.java | 41 +++++++--
.../computer/AbstractHadoopGraphComputer.java | 2 +-
.../hadoop/structure/io/GraphFilterAware.java | 17 ----
.../structure/io/gryo/GryoRecordReader.java | 4 +-
.../structure/io/script/ScriptRecordReader.java | 3 +-
.../spark/process/computer/SparkExecutor.java | 3 +-
.../process/computer/SparkGraphComputer.java | 3 +-
.../spark/structure/io/PersistedOutputRDD.java | 3 +-
.../process/computer/TinkerGraphComputer.java | 2 +-
14 files changed, 187 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexInputFormat.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexInputFormat.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexInputFormat.java
index 76fe3ba..3bb306b 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexInputFormat.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexInputFormat.java
@@ -45,7 +45,7 @@ public final class GiraphVertexInputFormat extends VertexInputFormat {
private InputFormat<NullWritable, VertexWritable> hadoopGraphInputFormat;
protected GraphFilter graphFilter = new GraphFilter();
private boolean graphFilterLoaded = false;
- private boolean graphFilterAware;
+ private boolean graphFilterAware = false;
@Override
public void checkInputSpecs(final Configuration configuration) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexReader.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexReader.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexReader.java
index d67937c..9780ca2 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexReader.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/structure/io/GiraphVertexReader.java
@@ -25,7 +25,6 @@ import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.tinkerpop.gremlin.giraph.process.computer.GiraphVertex;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.GraphFilterAware;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.process.computer.GraphFilter;
@@ -59,7 +58,7 @@ public final class GiraphVertexReader extends VertexReader {
while (true) {
if (this.recordReader.nextKeyValue()) {
final VertexWritable vertexWritable = this.recordReader.getCurrentValue();
- final org.apache.tinkerpop.gremlin.structure.Vertex vertex = GraphFilterAware.applyGraphFilter(vertexWritable.get(), this.graphFilter);
+ final org.apache.tinkerpop.gremlin.structure.Vertex vertex = this.graphFilter.applyGraphFilter(vertexWritable.get());
if (null != vertex) {
vertexWritable.set(vertex);
return true;
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
index 1156978..1494c28 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
@@ -109,7 +109,7 @@ public interface GraphComputer {
public GraphComputer vertices(final Traversal<Vertex, Vertex> vertexFilter);
- public GraphComputer edges(final Traversal<Edge, Edge> edgeFilter);
+ public GraphComputer edges(final Traversal<Vertex, Edge> edgeFilter);
/**
* Set an arbitrary configuration key/value for the underlying {@link org.apache.commons.configuration.Configuration} in the {@link GraphComputer}.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphFilter.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphFilter.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphFilter.java
index ea2080a..54256aa 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphFilter.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphFilter.java
@@ -20,10 +20,21 @@
package org.apache.tinkerpop.gremlin.process.computer;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.VertexStep;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
+import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.util.star.StarGraph;
import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -31,34 +42,43 @@ import java.io.Serializable;
public final class GraphFilter implements Cloneable, Serializable {
private Traversal.Admin<Vertex, Vertex> vertexFilter = null;
- private Traversal.Admin<Edge, Edge> edgeFilter = null;
+ private Traversal.Admin<Vertex, Edge> edgeFilter = null;
+
+ private Direction allowedEdgeDirection = Direction.BOTH;
+ private Set<String> allowedEdgeLabels = new HashSet<>();
+ private boolean allowAllRemainingEdges = false;
public void setVertexFilter(final Traversal<Vertex, Vertex> vertexFilter) {
this.vertexFilter = vertexFilter.asAdmin().clone();
}
- public void setEdgeFilter(final Traversal<Edge, Edge> edgeFilter) {
+ public void setEdgeFilter(final Traversal<Vertex, Edge> edgeFilter) {
this.edgeFilter = edgeFilter.asAdmin().clone();
+ if (this.edgeFilter.getStartStep() instanceof VertexStep) {
+ this.allowedEdgeLabels.addAll(Arrays.asList(((VertexStep) this.edgeFilter.getStartStep()).getEdgeLabels()));
+ this.allowedEdgeDirection = ((VertexStep) this.edgeFilter.getStartStep()).getDirection();
+ this.allowAllRemainingEdges = 1 == this.edgeFilter.getSteps().size();
+ }
}
- public boolean hasVertexFilter() {
- return this.vertexFilter != null;
+ public final Traversal.Admin<Vertex, Vertex> getVertexFilter() {
+ return this.vertexFilter.clone();
}
- public final Traversal.Admin<Vertex, Vertex> getVertexFilter() {
- return this.vertexFilter;
+ public final Traversal.Admin<Vertex, Edge> getEdgeFilter() {
+ return this.edgeFilter.clone();
}
- public final Traversal.Admin<Edge, Edge> getEdgeFilter() {
- return this.edgeFilter;
+ public boolean hasFilter() {
+ return this.vertexFilter != null || this.edgeFilter != null;
}
public boolean hasEdgeFilter() {
return this.edgeFilter != null;
}
- public boolean hasFilter() {
- return this.vertexFilter != null || this.edgeFilter != null;
+ public boolean hasVertexFilter() {
+ return this.vertexFilter != null;
}
@Override
@@ -86,4 +106,58 @@ public final class GraphFilter implements Cloneable, Serializable {
else
return "graphfilter[" + this.edgeFilter + "]";
}
+
+ //////////////////////////////////////
+ /////////////////////////////////////
+ ////////////////////////////////////
+
+ public StarGraph.StarVertex applyGraphFilter(final StarGraph.StarVertex vertex) {
+ if (!this.hasFilter())
+ return vertex;
+ else if (null == vertex)
+ return null;
+ else if (!this.hasVertexFilter() || TraversalUtil.test(vertex, this.vertexFilter)) {
+ if (this.hasEdgeFilter()) {
+ if (!this.allowedEdgeDirection.equals(Direction.BOTH))
+ vertex.dropEdges(this.allowedEdgeDirection.opposite());
+ if (!this.allowedEdgeLabels.isEmpty())
+ vertex.keepEdges(this.allowedEdgeDirection, this.allowedEdgeLabels);
+
+ if (!this.allowAllRemainingEdges) {
+ final Map<String, List<Edge>> outEdges = new HashMap<>();
+ final Map<String, List<Edge>> inEdges = new HashMap<>();
+ TraversalUtil.applyAll(vertex, this.edgeFilter).forEachRemaining(edge -> {
+ if (edge instanceof StarGraph.StarOutEdge) {
+ List<Edge> edges = outEdges.get(edge.label());
+ if (null == edges) {
+ edges = new ArrayList<>();
+ outEdges.put(edge.label(), edges);
+ }
+ edges.add(edge);
+ } else {
+ List<Edge> edges = inEdges.get(edge.label());
+ if (null == edges) {
+ edges = new ArrayList<>();
+ inEdges.put(edge.label(), edges);
+ }
+ edges.add(edge);
+ }
+ });
+
+ if (outEdges.isEmpty())
+ vertex.dropEdges(Direction.OUT);
+ else
+ vertex.setEdges(Direction.OUT, outEdges);
+
+ if (inEdges.isEmpty())
+ vertex.dropEdges(Direction.IN);
+ else
+ vertex.setEdges(Direction.IN, inEdges);
+ }
+ }
+ return vertex;
+ } else {
+ return null;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java
index c613dbb..5973df6 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java
@@ -39,10 +39,12 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
+import java.util.Set;
import java.util.stream.Stream;
/**
@@ -273,11 +275,57 @@ public final class StarGraph implements Graph, Serializable {
super(id, label);
}
- public void dropEdges() {
- if (null != this.outEdges) this.outEdges.clear();
- if (null != this.inEdges) this.inEdges.clear();
- this.outEdges = null;
- this.inEdges = null;
+ public void setEdges(final Direction direction, final Map<String, List<Edge>> edges) {
+ if (direction.equals(Direction.OUT))
+ this.outEdges = edges;
+ else if (direction.equals(Direction.IN))
+ this.inEdges = edges;
+ else
+ throw new IllegalArgumentException("The following direction is not supported: " + direction);
+ }
+
+ public void keepEdges(final Direction direction, final Set<String> edgeLabels) {
+ final Set<String> dropLabels = new HashSet<>();
+ if ((direction.equals(Direction.OUT) || direction.equals(Direction.BOTH)) && null != this.outEdges)
+ dropLabels.addAll(this.outEdges.keySet());
+ if ((direction.equals(Direction.IN) || direction.equals(Direction.BOTH)) && null != this.inEdges)
+ dropLabels.addAll(this.inEdges.keySet());
+ //
+ for (final String label : edgeLabels) {
+ dropLabels.remove(label);
+ }
+ if (dropLabels.size() > 0) {
+ this.dropEdges(direction, dropLabels);
+ }
+ }
+
+ public void dropEdges(final Direction direction) {
+ if ((direction.equals(Direction.OUT) || direction.equals(Direction.BOTH)) && null != this.outEdges) {
+ this.outEdges.clear();
+ this.outEdges = null;
+ }
+ if ((direction.equals(Direction.IN) || direction.equals(Direction.BOTH)) && null != this.inEdges) {
+ this.inEdges.clear();
+ this.inEdges = null;
+ }
+ }
+
+ public void dropEdges(final Direction direction, final Set<String> edgeLabels) {
+ if ((direction.equals(Direction.OUT) || direction.equals(Direction.BOTH)) && null != this.outEdges) {
+ for (final String edgeLabel : edgeLabels) {
+ this.outEdges.remove(edgeLabel);
+ }
+ if (this.outEdges.isEmpty())
+ this.outEdges = null;
+ }
+ if ((direction.equals(Direction.IN) || direction.equals(Direction.BOTH)) && null != this.inEdges) {
+ for (final String edgeLabel : edgeLabels) {
+ this.inEdges.remove(edgeLabel);
+ }
+
+ if (this.inEdges.isEmpty())
+ this.inEdges = null;
+ }
}
public void dropVertexProperties(final String... propertyKeys) {
@@ -441,7 +489,7 @@ public final class StarGraph implements Graph, Serializable {
@Override
public void remove() {
if (null != StarGraph.this.starVertex.vertexProperties)
- StarGraph.this.starVertex.vertexProperties.get(this.label()).remove(this);
+ StarGraph.this.starVertex.vertexProperties.get(this.label).remove(this);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
index 12c1bf3..c77468b 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
@@ -141,7 +141,7 @@ public class GraphComputerTest extends AbstractGremlinProcessTest {
}
@Override
- public GraphComputer edges(final Traversal<Edge, Edge> edgeFilter) {
+ public GraphComputer edges(final Traversal<Vertex, Edge> edgeFilter) {
return null;
}
@@ -1394,6 +1394,7 @@ public class GraphComputerTest extends AbstractGremlinProcessTest {
/////////////////////////////////////////////
@Test
+ @Ignore
@LoadGraphWith(GRATEFUL)
public void shouldSupportWorkerCount() throws Exception {
int maxWorkers = graph.compute(graphComputerClass.get()).features().getMaxWorkers();
@@ -1486,12 +1487,15 @@ public class GraphComputerTest extends AbstractGremlinProcessTest {
@Test
@LoadGraphWith(MODERN)
- public void shouldSupportVertexAndEdgeFilters() throws Exception {
+ public void shouldSupportGraphFilter() throws Exception {
graph.compute(graphComputerClass.get()).vertices(__.hasLabel("software")).program(new VertexProgramM(VertexProgramM.SOFTWARE_ONLY)).submit().get();
graph.compute(graphComputerClass.get()).vertices(__.hasLabel("person")).program(new VertexProgramM(VertexProgramM.PEOPLE_ONLY)).submit().get();
- graph.compute(graphComputerClass.get()).edges(__.hasLabel("knows")).program(new VertexProgramM(VertexProgramM.KNOWS_ONLY)).submit().get();
- graph.compute(graphComputerClass.get()).vertices(__.hasLabel("person")).edges(__.hasLabel("knows")).program(new VertexProgramM(VertexProgramM.PEOPLE_KNOWS_ONLY)).submit().get();
- graph.compute(graphComputerClass.get()).vertices(__.hasLabel("person")).edges(__.<Edge>hasLabel("knows").has("weight", P.gt(0.5f))).program(new VertexProgramM(VertexProgramM.PEOPLE_KNOWS_WELL_ONLY)).submit().get();
+ graph.compute(graphComputerClass.get()).edges(__.bothE("knows")).program(new VertexProgramM(VertexProgramM.KNOWS_ONLY)).submit().get();
+ graph.compute(graphComputerClass.get()).vertices(__.hasLabel("person")).edges(__.bothE("knows")).program(new VertexProgramM(VertexProgramM.PEOPLE_KNOWS_ONLY)).submit().get();
+ graph.compute(graphComputerClass.get()).vertices(__.hasLabel("person")).edges(__.<Vertex>bothE("knows").has("weight", P.gt(0.5f))).program(new VertexProgramM(VertexProgramM.PEOPLE_KNOWS_WELL_ONLY)).submit().get();
+ graph.compute(graphComputerClass.get()).edges(__.<Vertex>bothE().limit(0)).program(new VertexProgramM(VertexProgramM.VERTICES_ONLY)).submit().get();
+ graph.compute(graphComputerClass.get()).edges(__.<Vertex>outE().limit(1)).program(new VertexProgramM(VertexProgramM.ONE_OUT_EDGE_ONLY)).submit().get();
+ graph.compute(graphComputerClass.get()).edges(__.outE()).program(new VertexProgramM(VertexProgramM.OUT_EDGES_ONLY)).submit().get();
}
public static class VertexProgramM implements VertexProgram {
@@ -1501,6 +1505,9 @@ public class GraphComputerTest extends AbstractGremlinProcessTest {
public static final String KNOWS_ONLY = "knowsOnly";
public static final String PEOPLE_KNOWS_ONLY = "peopleKnowsOnly";
public static final String PEOPLE_KNOWS_WELL_ONLY = "peopleKnowsWellOnly";
+ public static final String VERTICES_ONLY = "verticesOnly";
+ public static final String ONE_OUT_EDGE_ONLY = "oneOutEdgeOnly";
+ public static final String OUT_EDGES_ONLY = "outEdgesOnly";
private String state;
@@ -1580,6 +1587,30 @@ public class GraphComputerTest extends AbstractGremlinProcessTest {
}
break;
}
+ case VERTICES_ONLY: {
+ assertEquals(0, IteratorUtils.count(vertex.edges(Direction.BOTH)));
+ break;
+ }
+ case ONE_OUT_EDGE_ONLY: {
+ if (vertex.label().equals("software") || vertex.value("name").equals("vadas"))
+ assertEquals(0, IteratorUtils.count(vertex.edges(Direction.BOTH)));
+ else {
+ assertEquals(1, IteratorUtils.count(vertex.edges(Direction.OUT)));
+ assertEquals(0, IteratorUtils.count(vertex.edges(Direction.IN)));
+ assertEquals(1, IteratorUtils.count(vertex.edges(Direction.BOTH)));
+ }
+ break;
+ }
+ case OUT_EDGES_ONLY: {
+ if (vertex.label().equals("software") || vertex.value("name").equals("vadas"))
+ assertEquals(0, IteratorUtils.count(vertex.edges(Direction.BOTH)));
+ else {
+ assertTrue(IteratorUtils.count(vertex.edges(Direction.OUT)) > 0);
+ assertEquals(0, IteratorUtils.count(vertex.edges(Direction.IN)));
+ assertEquals(IteratorUtils.count(vertex.edges(Direction.OUT)), IteratorUtils.count(vertex.edges(Direction.BOTH)));
+ }
+ break;
+ }
default:
throw new IllegalStateException("This is an illegal state for this test case: " + this.state);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
index a25cebd..ac83dd6 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
@@ -70,7 +70,7 @@ public abstract class AbstractHadoopGraphComputer implements GraphComputer {
}
@Override
- public GraphComputer edges(final Traversal<Edge, Edge> edgeFilter) {
+ public GraphComputer edges(final Traversal<Vertex, Edge> edgeFilter) {
this.graphFilter.setEdgeFilter(edgeFilter);
return this;
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/GraphFilterAware.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/GraphFilterAware.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/GraphFilterAware.java
index afc0f29..fa59a4e 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/GraphFilterAware.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/GraphFilterAware.java
@@ -42,23 +42,6 @@ public interface GraphFilterAware {
public void setGraphFilter(final GraphFilter graphFilter);
- public static Vertex applyGraphFilter(final Vertex vertex, final GraphFilter graphFilter) {
- if (null == vertex)
- return null;
- else if (!graphFilter.hasVertexFilter() || TraversalUtil.test(vertex, graphFilter.getVertexFilter())) {
- if (graphFilter.hasEdgeFilter()) {
- final Iterator<Edge> edgeIterator = vertex.edges(Direction.BOTH);
- while (edgeIterator.hasNext()) {
- if (!TraversalUtil.test(edgeIterator.next(), graphFilter.getEdgeFilter()))
- edgeIterator.remove();
- }
- }
- return vertex;
- } else {
- return null;
- }
- }
-
public static void storeGraphFilter(final Configuration apacheConfiguration, final org.apache.hadoop.conf.Configuration hadoopConfiguration, final GraphFilter graphFilter) {
if (graphFilter.hasFilter()) {
VertexProgramHelper.serialize(graphFilter, apacheConfiguration, Constants.GREMLIN_HADOOP_GRAPH_FILTER);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java
index 7339411..e489a15 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java
@@ -27,7 +27,6 @@ import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.GraphFilterAware;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.HadoopPools;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.process.computer.GraphFilter;
@@ -36,6 +35,7 @@ import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoMapper;
import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoReader;
import org.apache.tinkerpop.gremlin.structure.io.gryo.VertexTerminator;
import org.apache.tinkerpop.gremlin.structure.util.Attachable;
+import org.apache.tinkerpop.gremlin.structure.util.star.StarGraph;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@@ -128,7 +128,7 @@ public final class GryoRecordReader extends RecordReader<NullWritable, VertexWri
terminatorLocation = ((byte) currentByte) == TERMINATOR[terminatorLocation] ? terminatorLocation + 1 : 0;
if (terminatorLocation >= TERMINATOR.length) {
try (InputStream in = new ByteArrayInputStream(output.toByteArray())) {
- final Vertex vertex = GraphFilterAware.applyGraphFilter(this.gryoReader.readVertex(in, Attachable::get), this.graphFilter); // I know how GryoReader works, so I'm cheating here
+ final Vertex vertex = this.graphFilter.applyGraphFilter((StarGraph.StarVertex) this.gryoReader.readVertex(in, Attachable::get)); // I know how GryoReader works, so I'm cheating here
if (null != vertex) {
this.vertexWritable.set(vertex);
return true;
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReader.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReader.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReader.java
index c9b36b5..305cbd2 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReader.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReader.java
@@ -29,7 +29,6 @@ import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
import org.apache.tinkerpop.gremlin.groovy.CompilerCustomizerProvider;
import org.apache.tinkerpop.gremlin.groovy.DefaultImportCustomizerProvider;
import org.apache.tinkerpop.gremlin.groovy.jsr223.GremlinGroovyScriptEngine;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.GraphFilterAware;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.process.computer.GraphFilter;
import org.apache.tinkerpop.gremlin.structure.Edge;
@@ -88,7 +87,7 @@ public final class ScriptRecordReader extends RecordReader<NullWritable, VertexW
final Bindings bindings = this.engine.createBindings();
bindings.put(LINE, this.lineRecordReader.getCurrentValue().toString());
bindings.put(FACTORY, new ScriptElementFactory());
- final Vertex vertex = GraphFilterAware.applyGraphFilter((Vertex) engine.eval(READ_CALL, bindings), this.graphFilter);
+ final Vertex vertex = this.graphFilter.applyGraphFilter((StarGraph.StarVertex) engine.eval(READ_CALL, bindings));
if (vertex != null) {
this.vertexWritable.set(vertex);
return true;
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java
index ba80fbc..58a33ca 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java
@@ -22,7 +22,6 @@ import com.google.common.base.Optional;
import org.apache.commons.configuration.Configuration;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.GraphFilterAware;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.HadoopPools;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.process.computer.GraphFilter;
@@ -63,7 +62,7 @@ public final class SparkExecutor {
public static JavaPairRDD<Object, VertexWritable> filterLoadedGraph(JavaPairRDD<Object, VertexWritable> graphRDD, final GraphFilter graphFilter) {
return graphRDD.mapPartitionsToPair(partitionIterator -> {
final GraphFilter gFilter = graphFilter.clone();
- return () -> IteratorUtils.filter(partitionIterator, tuple -> GraphFilterAware.applyGraphFilter(tuple._2().get(), gFilter) != null);
+ return () -> IteratorUtils.filter(partitionIterator, tuple -> gFilter.applyGraphFilter(tuple._2().get()) != null);
}, true);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/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 1b59dd7..4b035e8 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
@@ -61,6 +61,7 @@ import org.apache.tinkerpop.gremlin.spark.structure.io.OutputRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedInputRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedOutputRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.SparkContextStorage;
+import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.io.Storage;
import java.io.File;
@@ -264,7 +265,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
if (computedGraphCreated && !outputToSpark) {
// drop all the edges of the graph as they are not used in mapReduce processing
computedGraphRDD = computedGraphRDD.mapValues(vertexWritable -> {
- vertexWritable.get().dropEdges();
+ vertexWritable.get().dropEdges(Direction.BOTH);
return vertexWritable;
});
// if there is only one MapReduce to execute, don't bother wasting the clock cycles.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java
index 4ae6248..0474c0c 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java
@@ -26,6 +26,7 @@ import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
import org.apache.tinkerpop.gremlin.spark.structure.Spark;
+import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,7 +51,7 @@ public final class PersistedOutputRDD implements OutputRDD {
final StorageLevel storageLevel = StorageLevel.fromString(configuration.getString(Constants.GREMLIN_SPARK_PERSIST_STORAGE_LEVEL, "MEMORY_ONLY"));
if (!configuration.getBoolean(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT_HAS_EDGES, true))
graphRDD.mapValues(vertex -> {
- vertex.get().dropEdges();
+ vertex.get().dropEdges(Direction.BOTH);
return vertex;
}).setName(Constants.getGraphLocation(configuration.getString(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION))).persist(storageLevel);
else
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d0ac6527/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
index 13f4fa3..f354f9f 100644
--- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
+++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
@@ -99,7 +99,7 @@ public final class TinkerGraphComputer implements GraphComputer {
}
@Override
- public GraphComputer edges(final Traversal<Edge, Edge> edgeFilter) {
+ public GraphComputer edges(final Traversal<Vertex, Edge> edgeFilter) {
throw new UnsupportedOperationException();
}