This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch csr
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/csr by this push:
new 1b08cffe1a cgc updates
1b08cffe1a is described below
commit 1b08cffe1afa9bc9d73154e500ad913746990e62
Author: Ken Hu <[email protected]>
AuthorDate: Thu Oct 8 09:35:48 2026 -0700
cgc updates
---
.../structure/snapshot/graph/CsrGraphComputer.java | 36 +++++--------
.../structure/snapshot/process/op/CsrOp.java | 8 +--
.../structure/snapshot/process/op/CsrPlan.java | 30 +++++++----
.../gremlin/structure/snapshot/process/op/Ops.java | 7 +--
.../structure/snapshot/process/op/Sources.java | 1 -
.../structure/snapshot/process/op/Terminals.java | 13 ++---
.../snapshot/process/op/value/MapOperators.java | 60 ++++++++++++++++++++++
.../snapshot/process/op/value/MapOps.java | 52 +++++++++++++++++--
.../process/op/value/ProjectKeyReader.java | 3 +-
.../snapshot/process/op/value/ValueOperators.java | 10 +++-
.../snapshot/process/strategy/PlanBuilder.java | 24 ++++-----
.../snapshot/process/strategy/Planner.java | 2 +-
.../snapshot/process/strategy/StepCompiler.java | 34 +++++++++---
spark-gremlin/src/test/csr-memory/csrbench.py | 2 +-
.../tinkerpop/gremlin/spark/csr/CsrMemoryRun.java | 21 +++++---
.../tinkerpop/gremlin/spark/csr/csr-queries.txt | 34 +++++++-----
16 files changed, 245 insertions(+), 92 deletions(-)
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/graph/CsrGraphComputer.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/graph/CsrGraphComputer.java
index 495bd98144..cf9cc8b5de 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/graph/CsrGraphComputer.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/graph/CsrGraphComputer.java
@@ -49,6 +49,7 @@ 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.PureTraversal;
import
org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
@@ -63,6 +64,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -111,12 +113,6 @@ public final class CsrGraphComputer implements
GraphComputer {
private static final String SHORTEST_DISTANCE =
"gremlin.shortestPathVertexProgram.distanceTraversal";
private static final String SHORTEST_MAX_DISTANCE =
"gremlin.shortestPathVertexProgram.maxDistance";
private static final String SHORTEST_INCLUDE_EDGES =
"gremlin.shortestPathVertexProgram.includeEdges";
- private static final String TRAVERSAL_VOTE_TO_HALT =
"gremlin.traversalVertexProgram.voteToHalt";
- private static final String TRAVERSAL_MUTATED_MEMORY_KEYS =
- "gremlin.traversalVertexProgram.mutatedMemoryKeys";
- private static final String TRAVERSAL_COMPLETED_BARRIERS =
- "gremlin.traversalVertexProgram.completedBarriers";
-
private final CsrGraph graph;
private final ExecutorService computerService =
Executors.newSingleThreadExecutor(r -> {
final Thread thread = new Thread(r,
CsrGraphComputer.class.getSimpleName() + "-boss");
@@ -192,9 +188,8 @@ public final class CsrGraphComputer implements
GraphComputer {
}
GraphComputerHelper.validateProgramOnComputer(this, vertexProgram);
- if (!vertexProgram.getMapReducers().isEmpty())
+ if (!(vertexProgram instanceof TraversalVertexProgram) &&
!vertexProgram.getMapReducers().isEmpty())
throw new UnsupportedOperationException("CsrGraphComputer does not
support VertexProgram MapReduce jobs");
- if (vertexProgram instanceof TraversalVertexProgram)
validateTraversalContinuation(vertexProgram);
resultGraph =
GraphComputerHelper.getResultGraphState(Optional.of(vertexProgram),
Optional.ofNullable(resultGraph));
@@ -305,20 +300,6 @@ public final class CsrGraphComputer implements
GraphComputer {
|| program instanceof TraversalVertexProgram;
}
- private static void validateTraversalContinuation(final VertexProgram<?>
program) {
- program.getMemoryComputeKeys().forEach(key -> {
- final String name = key.getKey();
- if (!name.equals(TraversalVertexProgram.HALTED_TRAVERSERS)
- && !name.equals(TraversalVertexProgram.ACTIVE_TRAVERSERS)
- && !name.equals(TRAVERSAL_VOTE_TO_HALT)
- && !name.equals(TRAVERSAL_MUTATED_MEMORY_KEYS)
- && !name.equals(TRAVERSAL_COMPLETED_BARRIERS)) {
- throw new UnsupportedOperationException("CsrGraphComputer does
not support traversal memory side "
- + "effects in TraversalVertexProgram: " + name);
- }
- });
- }
-
@Override
public Features features() {
return new Features() {
@@ -768,6 +749,7 @@ public final class CsrGraphComputer implements
GraphComputer {
final Traversal.Admin<Object, Object> traversal =
(Traversal.Admin<Object, Object>)
program.getTraversal().getPure();
traversal.setGraph(graph);
+ restoreGraphScans(traversal);
final boolean returnHaltedTraversers =
returnsHaltedTraversers(traversal);
if (!returnHaltedTraversers && !retainByVertex)
throw new UnsupportedOperationException("A chained
TraversalVertexProgram requires persisted vertex "
@@ -821,6 +803,16 @@ public final class CsrGraphComputer implements
GraphComputer {
return new TraversalResult(halted, haltedByVertex == null ? null :
new ObjectColumn(haltedByVertex));
}
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private void restoreGraphScans(final Traversal.Admin<?, ?> traversal) {
+ for (final GraphStep graphStep :
+
TraversalHelper.getStepsOfAssignableClassRecursively(GraphStep.class,
traversal)) {
+ graphStep.setIteratorSupplier(() -> (Iterator)
(graphStep.returnsVertex()
+ ? graph.vertices(graphStep.getIds())
+ : graph.edges(graphStep.getIds())));
+ }
+ }
+
private boolean returnsHaltedTraversers(final Traversal.Admin<?, ?>
traversal) {
final Step<?, ?> next =
traversal.getParent().asStep().getNextStep();
return next instanceof ComputerResultStep || next instanceof
EmptyStep
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/CsrOp.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/CsrOp.java
index 91d116675c..924b89e4ca 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/CsrOp.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/CsrOp.java
@@ -23,10 +23,12 @@ import
org.apache.tinkerpop.gremlin.structure.snapshot.process.exec.Lane;
import java.util.List;
/**
- * A node of the native operator IR, see the grammar in section 2.2 of the
spike document:
+ * A node of the native operator IR:
* <pre>
- * Plan := Source Op* Terminal?
+ * Plan := Source Node*
* </pre>
+ * A barrier {@link Terminal} consumes its whole input and may either end a
plan or feed later nodes. Reducing
+ * terminals emit one {@link Lane#SCALAR} entry; ordering terminals preserve
their input lane.
* Nodes are immutable descriptors built by the strategy; they hold no state
and no reference to a graph. The nodes are
* grouped in {@link Sources}, {@link Ops} and {@link Terminals}, with the key
forms in {@link Keys} and the predicate
* forms in {@link Preds}. Code that executes a node never switches on its
class: it is created through the
@@ -101,7 +103,7 @@ public interface CsrOp {
}
/**
- * A node that ends a plan: it consumes its whole input and emits a
reduced result, or nothing.
+ * A barrier node: it consumes its whole input before emitting results.
*/
interface Terminal extends CsrOp {
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/CsrPlan.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/CsrPlan.java
index 1332dcc61c..8fe6e75461 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/CsrPlan.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/CsrPlan.java
@@ -26,9 +26,10 @@ import java.util.List;
import java.util.Objects;
/**
- * An immutable operator plan: a {@link CsrOp.Source}, then any number of
ordinary nodes, then an optional
- * {@link CsrOp.Terminal}. Lanes are checked on construction by threading
{@link CsrOp#outputLane(Lane)} through the
- * nodes, so a plan that exists is lane-consistent. A plan that starts with
{@link Sources.Input} reads upstream
+ * An immutable operator plan: a {@link CsrOp.Source}, then any number of
nodes. A {@link CsrOp.Terminal} may be
+ * followed by ordinary nodes; a reducer's single {@link Lane#SCALAR} result
or an ordering terminal's original lane
+ * becomes the next node's input. Lanes are checked on construction by
threading {@link CsrOp#outputLane(Lane)} through
+ * the nodes, so a plan that exists is lane-consistent. A plan that starts
with {@link Sources.Input} reads upstream
* traversers (at the top level) or its parent's current batch (as a child).
*/
public final class CsrPlan {
@@ -38,7 +39,7 @@ public final class CsrPlan {
private final List<Boolean> sources;
/**
- * @param nodes the source, the middle nodes and optionally the terminal,
in order
+ * @param nodes the source and following nodes, in order
* @throws IllegalArgumentException if the shape or the lanes are
inconsistent
*/
public CsrPlan(final List<? extends CsrOp> nodes) {
@@ -57,9 +58,6 @@ public final class CsrPlan {
if (i > 0 && node instanceof CsrOp.Source && !(node instanceof
Sources.MidScan)) {
throw new IllegalArgumentException(node.name() + " can only
start a plan");
}
- if (node instanceof CsrOp.Terminal && i != this.nodes.size() - 1) {
- throw new IllegalArgumentException(node.name() + " can only
end a plan");
- }
if (node instanceof Ops.OtherV && !source) {
throw new IllegalArgumentException("OtherV needs edges that
recorded their source vertex");
}
@@ -88,14 +86,15 @@ public final class CsrPlan {
}
/**
- * The nodes between the source and the terminal.
+ * Every node after the source except an ending terminal. A terminal
followed by another node is included.
*/
public List<CsrOp> ops() {
return nodes.subList(1, terminal() == null ? nodes.size() :
nodes.size() - 1);
}
/**
- * The terminal, or null if the plan has none.
+ * The ending terminal, or null when the last node is not a terminal.
Intermediate terminals are available through
+ * {@link #ops()}.
*/
public CsrOp.Terminal terminal() {
final CsrOp last = nodes.get(nodes.size() - 1);
@@ -138,7 +137,18 @@ public final class CsrPlan {
}
/**
- * Whether the plan ends in a terminal.
+ * Whether the plan ends in one of the reducers supported as a {@code
group().by(valueTraversal)} value.
+ */
+ public boolean isGroupReducer() {
+ final CsrOp.Terminal terminal = terminal();
+ return terminal instanceof Terminals.Count || terminal instanceof
Terminals.Fold
+ || terminal instanceof Terminals.Sum || terminal instanceof
Terminals.Min
+ || terminal instanceof Terminals.Max || terminal instanceof
Terminals.Mean;
+ }
+
+ /**
+ * Whether the plan ends in any terminal/barrier. The historical method
name includes ordering terminals and drain,
+ * not only scalar reducers; use {@link #isGroupReducer()} when validating
a group value plan.
*/
public boolean isReducing() {
return terminal() != null;
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Ops.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Ops.java
index 6cac9e5373..921caed99e 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Ops.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Ops.java
@@ -912,8 +912,9 @@ public final class Ops {
Objects.requireNonNull(sideEffectKey);
Objects.requireNonNull(key);
Objects.requireNonNull(reducer);
- if (reducer.inputLane() == null || !reducer.isReducing()) {
- throw new IllegalArgumentException("A group reducer must start
with Input and end in a terminal");
+ if (reducer.inputLane() == null || !reducer.isGroupReducer()) {
+ throw new IllegalArgumentException("A group reducer must start
with Input and end in count, fold, "
+ + "sum, min, max or mean");
}
}
@@ -956,6 +957,6 @@ public final class Ops {
for (final CsrOp op : plan.ops()) {
if (!op.isBulkLinear()) return false;
}
- return !plan.isReducing();
+ return plan.terminal() == null;
}
}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Sources.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Sources.java
index 3588bbde34..740310b826 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Sources.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Sources.java
@@ -110,7 +110,6 @@ public final class Sources {
public record Input(Lane lane) implements CsrOp.Source {
public Input {
Objects.requireNonNull(lane);
- if (lane == Lane.SCALAR) throw new IllegalArgumentException("Input
cannot read SCALAR");
}
@Override
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Terminals.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Terminals.java
index d35a2dda6c..96cd4e75f3 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Terminals.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/Terminals.java
@@ -27,10 +27,10 @@ import java.util.List;
import java.util.Objects;
/**
- * The nodes that end a plan: {@code Terminal := Count | Sum | Min | Max |
Mean | GroupCount(key) | Group(key, Reducer) |
- * TopK(keys[], orders[], n, lo) | Sort(keys[], orders[]) | Fold | Drain}. A
terminal consumes its whole input. Those
- * that emit one result emit it as a single {@code SCALAR} entry of bulk 1,
and emit nothing for an empty input where
- * the standard step does (sum, min, max, mean).
+ * The barrier nodes: {@code Terminal := Count | Sum | Min | Max | Mean |
GroupCount(key) | Group(key, Reducer) |
+ * TopK(keys[], orders[], n, lo) | Sort(keys[], orders[]) | Fold | Drain}. A
terminal consumes its whole input before
+ * emitting results and may feed another node. Reducers emit one {@code
SCALAR} entry of bulk 1, and emit nothing for
+ * an empty input where the standard step does (sum, min, max, mean).
*/
public final class Terminals {
@@ -156,8 +156,9 @@ public final class Terminals {
public Group {
Objects.requireNonNull(key);
Objects.requireNonNull(reducer);
- if (reducer.inputLane() == null || !reducer.isReducing()) {
- throw new IllegalArgumentException("A group reducer must start
with Input and end in a terminal");
+ if (reducer.inputLane() == null || !reducer.isGroupReducer()) {
+ throw new IllegalArgumentException("A group reducer must start
with Input and end in count, fold, "
+ + "sum, min, max or mean");
}
}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/MapOperators.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/MapOperators.java
index 16b947e95c..2e7603ea49 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/MapOperators.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/MapOperators.java
@@ -18,6 +18,8 @@
*/
package org.apache.tinkerpop.gremlin.structure.snapshot.process.op.value;
+import org.apache.tinkerpop.gremlin.process.traversal.Path;
+import org.apache.tinkerpop.gremlin.process.traversal.step.util.Tree;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.WithOptions;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.T;
@@ -25,15 +27,21 @@ import
org.apache.tinkerpop.gremlin.structure.snapshot.CsrSnapshot;
import org.apache.tinkerpop.gremlin.structure.snapshot.graph.CsrGraph;
import org.apache.tinkerpop.gremlin.structure.snapshot.process.exec.Batch;
import org.apache.tinkerpop.gremlin.structure.snapshot.process.exec.Lane;
+import
org.apache.tinkerpop.gremlin.structure.snapshot.process.exec.Materializer;
import
org.apache.tinkerpop.gremlin.structure.snapshot.process.exec.OperatorSpec;
import org.apache.tinkerpop.gremlin.structure.snapshot.process.op.Keys;
import
org.apache.tinkerpop.gremlin.structure.snapshot.process.op.nav.EntryStreamOperator;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.tinkerpop.gremlin.util.NumberHelper.max;
+
/**
* The operators behind {@link MapOps}. Each builds one {@code LinkedHashMap}
per input entry from the owner ranges and
* columns, in the key order of the standard step, and emits it as a decoded
{@code VAL} entry with the entry's bulk.
@@ -257,4 +265,56 @@ final class MapOperators {
readers = null;
}
}
+
+ static final class SelectColumnOperator extends EntryStreamOperator {
+ private final MapOps.SelectColumn node;
+
+ SelectColumnOperator(final MapOps.SelectColumn node, final
OperatorSpec spec) {
+ super(spec);
+ this.node = node;
+ }
+
+ @Override
+ protected boolean process(final Batch in, final int i, final Batch
out) {
+ out.addValue(node.column().apply(Materializer.value(ctx, in, i)),
in.bulk[i]);
+ return true;
+ }
+ }
+
+ static final class CountLocalOperator extends EntryStreamOperator {
+
+ CountLocalOperator(final OperatorSpec spec) {
+ super(spec);
+ }
+
+ @Override
+ protected boolean process(final Batch in, final int i, final Batch
out) {
+ final Object item = Materializer.materialize(ctx, in, i);
+ final long count = item instanceof Tree ? ((Tree) item).nodeCount()
+ : item instanceof Collection ? ((Collection<?>)
item).size()
+ : item instanceof Map ? ((Map<?, ?>) item).size()
+ : item instanceof Path ? ((Path) item).size()
+ : IteratorUtils.count(IteratorUtils.asIterator(item));
+ out.addValue(count, in.bulk[i]);
+ return true;
+ }
+ }
+
+ static final class MaxLocalOperator extends EntryStreamOperator {
+
+ MaxLocalOperator(final OperatorSpec spec) {
+ super(spec);
+ }
+
+ @Override
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ protected boolean process(final Batch in, final int i, final Batch
out) {
+ final Iterator<?> iterator =
IteratorUtils.asIterator(Materializer.materialize(ctx, in, i));
+ if (!iterator.hasNext()) return true;
+ Comparable result = (Comparable) iterator.next();
+ while (iterator.hasNext()) result = max((Comparable)
iterator.next(), result);
+ out.addValue(result, in.bulk[i]);
+ return true;
+ }
+ }
}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/MapOps.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/MapOps.java
index 8e7215ecd7..fa3aea6e22 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/MapOps.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/MapOps.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.structure.snapshot.process.op.value;
+import org.apache.tinkerpop.gremlin.structure.Column;
import org.apache.tinkerpop.gremlin.structure.snapshot.process.exec.Lane;
import org.apache.tinkerpop.gremlin.structure.snapshot.process.op.CsrOp;
import org.apache.tinkerpop.gremlin.structure.snapshot.process.op.CsrPlan;
@@ -31,10 +32,10 @@ import java.util.Objects;
/**
* The IR nodes of the native map emission, which the shared IR in {@code Ops}
has no form for: {@code valueMap()},
- * {@code propertyMap()}, {@code elementMap()} and {@code project()}. Each
node turns an entry into one map and emits it
- * on the {@code VAL} lane with the entry's bulk; the maps are built directly
from owner ranges and columns. Key codes
- * are vertex or edge key codes by the input lane; an empty array means every
key in key-code order, and negative codes
- * (keys the snapshot does not have) are skipped. The planner must only emit a
node after checking
+ * {@code propertyMap()}, {@code elementMap()}, {@code project()} and local
maps over decoded values. Each node emits
+ * on the {@code VAL} lane with the entry's bulk. Element maps are built
directly from owner ranges and columns. Key
+ * codes are vertex or edge key codes by the input lane; an empty array means
every key in key-code order, and negative
+ * codes (keys the snapshot does not have) are skipped. The planner must only
emit a node after checking
* {@code CsrOperatorFactory#isImplemented}.
*/
public final class MapOps {
@@ -141,7 +142,7 @@ public final class MapOps {
@Override
public Lane outputLane(final Lane input) {
- requireLane(this, input, Lane.V, Lane.E, Lane.VP, Lane.EP,
Lane.MP, Lane.VAL);
+ requireLane(this, input, Lane.V, Lane.E, Lane.VP, Lane.EP,
Lane.MP, Lane.VAL, Lane.SCALAR);
for (final Keys.Key key : keys) {
for (final CsrPlan child : key.children()) {
if (child.inputLane() != input) {
@@ -165,4 +166,45 @@ public final class MapOps {
return "Project(" + names + "," + keys + ")";
}
}
+
+ /**
+ * {@code select(keys)} or {@code select(values)} over a decoded {@link
java.util.Map},
+ * {@link java.util.Map.Entry} or {@link
org.apache.tinkerpop.gremlin.process.traversal.Path}. The standard
+ * {@link Column} implementation defines both the accepted inputs and the
returned collection/value.
+ */
+ public record SelectColumn(Column column) implements CsrOp {
+ public SelectColumn {
+ Objects.requireNonNull(column);
+ }
+
+ @Override
+ public Lane outputLane(final Lane input) {
+ requireLane(this, input, Lane.VAL, Lane.SCALAR);
+ return Lane.VAL;
+ }
+ }
+
+ /**
+ * {@code count(local)} over a decoded value. The operator follows {@code
CountLocalStep}: collections, maps and
+ * paths use their sizes, trees use their recursive node count, and other
values are viewed through
+ * {@code IteratorUtils.asIterator}.
+ */
+ public record CountLocal() implements CsrOp {
+ @Override
+ public Lane outputLane(final Lane input) {
+ Objects.requireNonNull(input);
+ return Lane.VAL;
+ }
+ }
+
+ /**
+ * {@code max(local)} over a decoded local value. Empty inputs are
unproductive, as in the standard step.
+ */
+ public record MaxLocal() implements CsrOp {
+ @Override
+ public Lane outputLane(final Lane input) {
+ Objects.requireNonNull(input);
+ return Lane.VAL;
+ }
+ }
}
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/ProjectKeyReader.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/ProjectKeyReader.java
index c599efbbed..018ea8864d 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/ProjectKeyReader.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/ProjectKeyReader.java
@@ -144,7 +144,8 @@ final class ProjectKeyReader implements AutoCloseable {
final Property<?> p = ((Element) Materializer.facade(ctx, b,
i)).property(k.name());
return p.isPresent() ? p.value() : absent(k);
}
- case VAL: {
+ case VAL:
+ case SCALAR: {
final Object v = Materializer.value(ctx, b, i);
if (v instanceof Map) return ((Map<?, ?>) v).get(k.name());
throw new IllegalStateException(String.format("The by(\"%s\")
modulator can only be applied to a "
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/ValueOperators.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/ValueOperators.java
index d7fbc767ca..344b45acaf 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/ValueOperators.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/op/value/ValueOperators.java
@@ -30,8 +30,9 @@ import
org.apache.tinkerpop.gremlin.structure.snapshot.process.op.Ops;
* <ul>
* <li>{@code Props}, {@code PropKey}, {@code PropValue}, {@code Id},
{@code Label}, {@code Labels},
* {@code Element} and {@code Constant} from the shared IR;</li>
- * <li>{@link MapOps.ValueMap}, {@link MapOps.PropertyMap}, {@link
MapOps.ElementMap} and {@link MapOps.Project},
- * nodes that live in this package.</li>
+ * <li>{@link MapOps.ValueMap}, {@link MapOps.PropertyMap}, {@link
MapOps.ElementMap}, {@link MapOps.Project},
+ * {@link MapOps.SelectColumn}, {@link MapOps.CountLocal} and {@link
MapOps.MaxLocal}, nodes that live in this
+ * package.</li>
* </ul>
*/
public final class ValueOperators implements OperatorRegistrar {
@@ -50,6 +51,11 @@ public final class ValueOperators implements
OperatorRegistrar {
factory.register(MapOps.PropertyMap.class,
MapOperators.PropertyMapOperator::new);
factory.register(MapOps.ElementMap.class,
MapOperators.ElementMapOperator::new);
factory.register(MapOps.Project.class,
MapOperators.ProjectOperator::new);
+ factory.register(MapOps.SelectColumn.class,
MapOperators.SelectColumnOperator::new);
+ factory.register(MapOps.CountLocal.class,
+ (node, spec) -> new MapOperators.CountLocalOperator(spec));
+ factory.register(MapOps.MaxLocal.class,
+ (node, spec) -> new MapOperators.MaxLocalOperator(spec));
factory.registerEstimator(Ops.Props.class, (node, snapshot, in) -> {
final double factor = node.keyCodes().length == 0 ?
allKeys(snapshot) : node.keyCodes().length;
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/PlanBuilder.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/PlanBuilder.java
index f965309f2d..de4fc3e5df 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/PlanBuilder.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/PlanBuilder.java
@@ -27,15 +27,16 @@ import java.util.ArrayList;
import java.util.List;
/**
- * The nodes of a plan under construction, with the lane after the last node.
A terminal closes the builder. Copies are
- * cheap, so a composite rule can build on a copy and commit it only when
every piece compiled.
+ * The nodes of a plan under construction, with the lane after the last node.
A compatible node may continue from a
+ * terminal's output. Copies are cheap, so a composite rule can build on a
copy and commit it only when every piece
+ * compiled.
*/
final class PlanBuilder {
private final List<CsrOp> nodes = new ArrayList<>();
private Lane lane;
private boolean recordsSource;
- private boolean closed;
+ private boolean hardClosed;
PlanBuilder(final CsrOp.Source source) {
add(source);
@@ -45,7 +46,7 @@ final class PlanBuilder {
this.nodes.addAll(other.nodes);
this.lane = other.lane;
this.recordsSource = other.recordsSource;
- this.closed = other.closed;
+ this.hardClosed = other.hardClosed;
}
PlanBuilder copy() {
@@ -60,16 +61,16 @@ final class PlanBuilder {
nodes.addAll(copy.nodes);
lane = copy.lane;
recordsSource = copy.recordsSource;
- closed = copy.closed;
+ hardClosed = copy.hardClosed;
}
/**
* Appends a node.
*
- * @throws Reject if the plan is closed or the node cannot read the
current lane
+ * @throws Reject if the plan was explicitly closed or the node cannot
read the current lane
*/
void add(final CsrOp node) {
- if (closed) throw new Reject("no step may follow the terminal " +
nodes.get(nodes.size() - 1).name());
+ if (hardClosed) throw new Reject("the plan cannot continue after a
side-effect writer");
try {
lane = node.outputLane(nodes.isEmpty() ? null : lane);
} catch (final IllegalArgumentException e) {
@@ -77,7 +78,6 @@ final class PlanBuilder {
}
recordsSource = lane == Lane.E &&
node.outputRecordsSource(recordsSource);
nodes.add(node);
- if (node instanceof CsrOp.Terminal) closed = true;
}
/**
@@ -86,7 +86,7 @@ final class PlanBuilder {
void replace(final int index, final CsrOp node) {
final List<CsrOp> old = new ArrayList<>(nodes);
nodes.clear();
- closed = false;
+ hardClosed = false;
for (int i = 0; i < old.size(); i++) add(i == index ? node :
old.get(i));
}
@@ -110,11 +110,11 @@ final class PlanBuilder {
* Ends the plan after the last node without a terminal.
*/
void close() {
- closed = true;
+ hardClosed = true;
}
- boolean isClosed() {
- return closed;
+ boolean endsInTerminal() {
+ return nodes.get(nodes.size() - 1) instanceof CsrOp.Terminal;
}
boolean hasOps() {
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/Planner.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/Planner.java
index 5e0b2ebb55..a74213398c 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/Planner.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/Planner.java
@@ -210,7 +210,7 @@ final class Planner {
}
}
j += used;
- if (close || b.isClosed()) break;
+ if (close) break;
} catch (final Reject r) {
b.commit(saved);
note(steps.get(j), j, r.getMessage());
diff --git
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/StepCompiler.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/StepCompiler.java
index a0cdd4ae99..0408a36465 100644
---
a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/StepCompiler.java
+++
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/snapshot/process/strategy/StepCompiler.java
@@ -24,6 +24,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.Pick;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import
org.apache.tinkerpop.gremlin.process.traversal.lambda.AbstractLambdaTraversal;
+import org.apache.tinkerpop.gremlin.process.traversal.lambda.ColumnTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.lambda.ConstantTraversal;
import
org.apache.tinkerpop.gremlin.process.traversal.lambda.GValueConstantTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.lambda.IdentityTraversal;
@@ -58,6 +59,7 @@ import
org.apache.tinkerpop.gremlin.process.traversal.step.filter.TraversalFilte
import org.apache.tinkerpop.gremlin.process.traversal.step.map.CoalesceStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.ConstantStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.CountGlobalStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.CountLocalStep;
import
org.apache.tinkerpop.gremlin.process.traversal.step.map.EdgeOtherVertexStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.EdgeVertexStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.ElementMapStep;
@@ -71,6 +73,7 @@ import
org.apache.tinkerpop.gremlin.process.traversal.step.map.LabelStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.LabelsStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.LoopsStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.MaxGlobalStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.MaxLocalStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.MeanGlobalStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.MinGlobalStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.NoOpBarrierStep;
@@ -442,6 +445,10 @@ final class StepCompiler {
b.add(new Terminals.Count());
return 1;
}
+ if (s instanceof CountLocalStep) {
+ b.add(new MapOps.CountLocal());
+ return 1;
+ }
if (s instanceof SumGlobalStep) {
requireValues(lane, "sum()");
b.add(new Terminals.Sum());
@@ -457,6 +464,10 @@ final class StepCompiler {
b.add(new Terminals.Max());
return 1;
}
+ if (s instanceof MaxLocalStep) {
+ b.add(new MapOps.MaxLocal());
+ return 1;
+ }
if (s instanceof MeanGlobalStep) {
requireValues(lane, "mean()");
b.add(new Terminals.Mean());
@@ -523,6 +534,10 @@ final class StepCompiler {
if (lane != Lane.VAL) throw new Reject(what + " needs values, not " +
lane);
}
+ private static void requireDecoded(final Lane lane, final String what) {
+ if (lane != Lane.VAL && lane != Lane.SCALAR) throw new Reject(what + "
needs a decoded value, not " + lane);
+ }
+
private static void requireMapLane(final Lane lane) {
if (lane != Lane.V && lane != Lane.E) throw new Reject("a map step
needs vertices or edges, not " + lane);
}
@@ -756,9 +771,10 @@ final class StepCompiler {
* wrapper), constants, or a child plan whose first result is the key.
*/
private Keys.Key key(final Traversal.Admin<?, ?> by, final Lane lane,
final Scope sc) {
- if (lane == null || lane == Lane.SCALAR) throw new Reject("a key over
" + lane);
+ if (lane == null) throw new Reject("a key over " + lane);
if (by == null || by instanceof IdentityTraversal) return new
Keys.Identity();
if (by instanceof TokenTraversal) {
+ if (lane == Lane.SCALAR) throw new Reject("by(token) over " +
lane);
final T token = ((TokenTraversal<?, ?>) by).getToken();
final boolean ok;
if (token == T.id) ok = lane == Lane.V || lane == Lane.E || lane
== Lane.VP;
@@ -772,9 +788,10 @@ final class StepCompiler {
final ValueTraversal<?, ?> vt = (ValueTraversal<?, ?>) by;
final boolean productive = vt.getBypassTraversal() != null;
if (productive && !isProductiveBypass(vt)) throw new Reject("an
unrecognized by() bypass");
+ final String name = vt.getPropertyKey();
+ if (lane == Lane.VAL || lane == Lane.SCALAR) return new
Keys.Value(-1, name, productive);
if (lane != Lane.V && lane != Lane.E) throw new Reject("by(key)
over " + lane);
requireFull("properties");
- final String name = vt.getPropertyKey();
final int code = lane == Lane.V ? graph.vertexKeyCode(name) :
graph.edgeKeyCode(name);
if (code < 0) {
if (productive) return new Keys.Const(null);
@@ -843,7 +860,7 @@ final class StepCompiler {
*/
private CsrPlan child(final Traversal.Admin<?, ?> t, final Lane lane,
final Scope sc, final boolean allowTerminal) {
if (t instanceof AbstractLambdaTraversal) throw new Reject("a lambda
traversal child");
- if (lane == null || lane == Lane.SCALAR) throw new Reject("a child
over " + lane);
+ if (lane == null) throw new Reject("a child over " + lane);
final CsrOp.Source input = new Sources.Input(lane);
checkImplemented(input);
final PlanBuilder b = new PlanBuilder(input);
@@ -859,7 +876,7 @@ final class StepCompiler {
if (analysis.anyReferenced(s.getLabels())) throw new Reject("the
label of " + name(s) + " is read");
i += compile(steps, i, b, inner);
}
- if (b.isClosed() && !allowTerminal) throw new Reject("a reducing
child");
+ if (b.endsInTerminal() && !allowTerminal) throw new Reject("a reducing
child");
return b.build();
}
@@ -874,6 +891,11 @@ final class StepCompiler {
if (lt.getBypassTraversal() != null) throw new Reject("a map over a
bypass traversal");
if (lt instanceof IdentityTraversal) return;
final Lane lane = b.lane();
+ if (lt instanceof ColumnTraversal) {
+ requireDecoded(lane, "select(column)");
+ b.add(new MapOps.SelectColumn(((ColumnTraversal) lt).getColumn()));
+ return;
+ }
if (lt instanceof TokenTraversal) {
final T token = ((TokenTraversal<?, ?>) lt).getToken();
if (token == T.id) {
@@ -926,7 +948,7 @@ final class StepCompiler {
private void project(final ProjectStep<?, ?> p, final PlanBuilder b, final
Scope sc) {
final Lane lane = b.lane();
- if (lane == null || lane == Lane.SCALAR) throw new Reject("project()
of " + lane);
+ if (lane == null) throw new Reject("project() of " + lane);
final List<String> names = p.getProjectKeys();
final List<? extends Traversal.Admin<?, ?>> ring =
p.getTraversalRing().getTraversals();
final List<Keys.Key> keys = new ArrayList<>(names.size());
@@ -1055,7 +1077,7 @@ final class StepCompiler {
if (analysis.anyReferenced(s.getLabels())) throw new Reject("the
label of " + name(s) + " is read");
i += compile(steps, i, b, inner);
}
- if (b.isClosed()) throw new Reject("a reducing loop child");
+ if (b.endsInTerminal()) throw new Reject("a reducing loop child");
return b.build();
}
diff --git a/spark-gremlin/src/test/csr-memory/csrbench.py
b/spark-gremlin/src/test/csr-memory/csrbench.py
index 2b846410a2..51bd4cc8b2 100755
--- a/spark-gremlin/src/test/csr-memory/csrbench.py
+++ b/spark-gremlin/src/test/csr-memory/csrbench.py
@@ -48,7 +48,7 @@ DEFAULT_QUERY_FILE = os.path.join(
DEFAULT_CLASSPATH_FILE = os.path.join(REPO_ROOT, "spark-gremlin", "target",
"csrbench-classpath.txt")
MB = 1 << 20
GB = 1 << 30
-CSR_SYSTEMS = ("csr", "csr-native", "csr-facade")
+CSR_SYSTEMS = ("csr", "csr-native", "csr-facade", "csr-computer")
ADD_OPENS = [
"java.base/java.util.concurrent.atomic", "java.base/java.util",
"java.base/java.lang",
diff --git
a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/csr/CsrMemoryRun.java
b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/csr/CsrMemoryRun.java
index 135f463f17..a65668bbe5 100644
---
a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/csr/CsrMemoryRun.java
+++
b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/csr/CsrMemoryRun.java
@@ -41,6 +41,7 @@ import
org.apache.tinkerpop.gremlin.structure.snapshot.build.HybridSnapshotBuild
import org.apache.tinkerpop.gremlin.structure.snapshot.build.SnapshotBuilder;
import
org.apache.tinkerpop.gremlin.structure.snapshot.build.StreamingSnapshotBuilder;
import org.apache.tinkerpop.gremlin.structure.snapshot.graph.CsrGraph;
+import org.apache.tinkerpop.gremlin.structure.snapshot.graph.CsrGraphComputer;
import org.apache.tinkerpop.gremlin.structure.snapshot.process.CsrNative;
import org.apache.tinkerpop.gremlin.structure.snapshot.process.CsrSuperStep;
import
org.apache.tinkerpop.gremlin.structure.snapshot.process.exec.CsrMemoryBudgetException;
@@ -82,7 +83,7 @@ import java.util.Set;
* <pre>
* java [jvm flags] org.apache.tinkerpop.gremlin.spark.csr.CsrMemoryRun
* --mode build|open|query
- * --system tinkergraph|csr-native|csr-facade|tinkergraph-computer|spark
(query mode; open mode: tinkergraph|csr)
+ * --system
tinkergraph|csr-native|csr-facade|tinkergraph-computer|spark|csr-computer
(query mode; open mode: tinkergraph|csr)
* --dataset <.kryo file> (required for tinkergraph*,
spark, and build)
* --snapshot <snapshot dir> (csr-* and open csr; build
writes here and it must not exist)
* --query-file <file> --query <id> (query mode; the file is
described below)
@@ -111,7 +112,9 @@ import java.util.Set;
* uses) and runs OLTP. {@code tinkergraph-computer} is the same graph with
* {@code g.withComputer()}. {@code csr-native} opens the snapshot and runs
with {@code csrMemoryBudget} set from
* {@code --csr-budget} and {@code csrScratchDirectory} under the work
directory; {@code csr-facade} adds
- * {@code withoutStrategies(CsrNativeStrategy)}. {@code spark} runs {@code
g.withComputer(SparkGraphComputer)} over a
+ * {@code withoutStrategies(CsrNativeStrategy)}. {@code csr-computer} opens
the snapshot and runs with
+ * {@code withComputer(CsrGraphComputer)}, which executes the vertex-program
steps below directly over the snapshot.
+ * {@code spark} runs {@code g.withComputer(SparkGraphComputer)} over a
* {@link HadoopGraph} that reads the Gryo file with {@code GryoInputFormat}
in one JVM ({@code spark.master}
* {@code local[*]} by default, the Kryo serializer with {@code
GryoRegistrator}, graph and persist storage levels
* {@code MEMORY_AND_DISK}, {@code spark.local.dir} and the Gremlin output
location under the work directory, the Spark
@@ -132,8 +135,9 @@ import java.util.Set;
* <p/>
* <b>Vertex-program steps.</b> {@code pageRank()}, {@code
connectedComponent()} and {@code shortestPath()}, with their
* {@code PageRank}, {@code ConnectedComponent} and {@code ShortestPath}
{@code with()} options, are part of
- * gremlin-lang, so such queries are ordinary lines that list only {@code
tinkergraph-computer} and {@code spark}
- * (CSR has no graph computer, and OLTP TinkerGraph rejects the steps); the
other systems report UNSUPPORTED. The
+ * gremlin-lang, and so is {@code peerPressure()} with its {@code
PeerPressure} options. Such queries are ordinary
+ * lines that list the graph computers {@code tinkergraph-computer}, {@code
spark} and {@code csr-computer}; OLTP
+ * TinkerGraph and the OLTP CSR systems reject the steps and report
UNSUPPORTED. The
* query reads what the program wrote ({@code values('pageRank')}, {@code
by('component')}, the paths) in the same
* traversal, so count and hash cover it. On Spark the steps chain jobs
through the Gryo graph writer and output
* location configured below.
@@ -170,11 +174,12 @@ public final class CsrMemoryRun {
private static final int SAMPLE_RESULTS = 10;
private static final int SAMPLE_CHARS = 2000;
- private static final Set<String> SYSTEMS = Set.of("tinkergraph",
"csr-native", "csr-facade", "tinkergraph-computer", "spark");
+ private static final Set<String> SYSTEMS = Set.of("tinkergraph",
"csr-native", "csr-facade", "tinkergraph-computer", "spark",
+ "csr-computer");
private static final String USAGE = String.join(System.lineSeparator(),
"Usage: CsrMemoryRun --mode build|open|query [options]",
- " --system
tinkergraph|csr-native|csr-facade|tinkergraph-computer|spark (open:
tinkergraph|csr)",
+ " --system
tinkergraph|csr-native|csr-facade|tinkergraph-computer|spark|csr-computer
(open: tinkergraph|csr)",
" --dataset <.kryo> --snapshot <dir> --query-file <file>
--query <id> --warm <n> (3)",
" --timeout-seconds <n> --csr-budget <bytes> (1073741824)
--builder heap|streaming|hybrid (streaming)",
" --source gryo|tinkergraph (gryo) --builder-budget <bytes>
(268435456)",
@@ -418,6 +423,10 @@ public final class CsrMemoryRun {
openCsr(false);
g =
csrGraph.traversal().withoutStrategies(CsrNativeStrategy.class);
break;
+ case "csr-computer":
+ openCsr(false);
+ g = csrGraph.traversal().withComputer(CsrGraphComputer.class);
+ break;
case "spark":
require(config.dataset, "--dataset");
hadoopGraph = HadoopGraph.open(sparkConfiguration());
diff --git
a/spark-gremlin/src/test/resources/org/apache/tinkerpop/gremlin/spark/csr/csr-queries.txt
b/spark-gremlin/src/test/resources/org/apache/tinkerpop/gremlin/spark/csr/csr-queries.txt
index 9240ae5261..3a7dc3d873 100644
---
a/spark-gremlin/src/test/resources/org/apache/tinkerpop/gremlin/spark/csr/csr-queries.txt
+++
b/spark-gremlin/src/test/resources/org/apache/tinkerpop/gremlin/spark/csr/csr-queries.txt
@@ -183,8 +183,11 @@ sid-groupcount-id | small | * |
g.V().groupCount().by(id).count(local)
#
---------------------------------------------------------------------------------------------------------------------
# Graph algorithms (alg-*), on the standard power-law graphs. Each algorithm
has a recipe written in plain Gremlin
# (docs/src/recipes), which runs on every system, and where TinkerPop has one
a vertex-program step (pageRank(),
-# connectedComponent(), shortestPath()), which only the graph computers can
run: the -vp queries list
-# tinkergraph-computer and spark, and the other systems report UNSUPPORTED.
Both forms end in the same summary (a
+# peerPressure(), connectedComponent(), shortestPath()), which only the graph
computers can run: the -vp queries list
+# tinkergraph-computer, spark and csr-computer (CsrGraphComputer), and the
other systems report UNSUPPORTED.
+# matrix-algorithms.json runs only the four vertex programs (pagerank,
peerpressure, cc, sp) on the three graph
+# computers; the recipes and the other -vp variants stay here for
matrix-algorithms-recipes.json, which is kept for
+# reference and not run. Both forms end in the same summary (a
# count, a min, a list of counts, the ids of a path), so that recipe and
vertex program can be compared by result hash; the exception is
# alg-eigen-pagerank-vp, a different centrality whose top ten is compared with
alg-eigen-walks by eye
# (extra.resultSample).
@@ -212,7 +215,12 @@ alg-degree-top10 | small | * | g.V().project('v',
'd').by(id).by(bothE().co
# alg: eigenvector centrality. The recipe counts out-walks of length 0-4 that
end at each vertex; barrier() bulks the
# traversers at each step, without it the walks are enumerated one by one.
pageRank() is the vertex-program reference.
alg-eigen-walks | small | * |
g.V().repeat(groupCount('m').by(id).out().barrier()).times(5).cap('m').order(local).by(values,
desc).limit(local, 10)
-alg-eigen-pagerank-vp | small | tinkergraph-computer,spark |
g.V().pageRank().with(PageRank.propertyName, 'pageRank').with(PageRank.times,
20).order().by('pageRank', desc).by(id).limit(10).id() | ordered
+alg-eigen-pagerank-vp | small | tinkergraph-computer,spark,csr-computer |
g.V().pageRank().with(PageRank.propertyName, 'pageRank').with(PageRank.times,
20).order().by('pageRank', desc).by(id).limit(10).id() | ordered
+
+# alg: label propagation clustering (peerPressure(), 20 iterations),
summarized as cluster count and largest cluster
+# size. It has no recipe. Ties in the vote go to the smaller cluster id by
string order, so the result is
+# deterministic, but vote sums are floating point and a different summation
order can still change a close vote.
+alg-peerpressure-vp | small | tinkergraph-computer,spark,csr-computer |
g.V().peerPressure().with(PeerPressure.propertyName,
'cluster').with(PeerPressure.times,
20).groupCount().by('cluster').select(values).project('clusters',
'largest').by(count(local)).by(max(local))
# alg: connected components (weak), summarized as component count and largest
component size. alg-cc-recipe runs a
# dedup() breadth-first search per vertex not yet seen, inside local(); the
graph computers reject local() beyond the
@@ -221,37 +229,37 @@ alg-eigen-pagerank-vp | small |
tinkergraph-computer,spark | g.V().pageRank().wi
# alg-cc-vp reads the default component key, because gremlin-lang does not
implement ConnectedComponent.propertyName.
alg-cc-recipe | small | * |
g.V().local(where(without('a')).union(identity(),
repeat(both().dedup()).emit()).dedup().aggregate('a').count()).is(gt(0)).fold().project('components',
'largest').by(count(local)).by(max(local))
alg-cc-recipe-doc | small | * | g.V().emit(or(cyclicPath(),
not(both()))).repeat(where(without('a')).local(aggregate('a')).both()).until(cyclicPath()).group().by(path().unfold().limit(1)).by(path().unfold().dedup().fold()).select(values).unfold().count(local).fold().project('components',
'largest').by(count(local)).by(max(local))
-alg-cc-vp | small | tinkergraph-computer,spark |
g.V().connectedComponent().groupCount().by('gremlin.connectedComponentVertexProgram.component').select(values).project('components',
'largest').by(count(local)).by(max(local))
+alg-cc-vp | small | tinkergraph-computer,spark,csr-computer |
g.V().connectedComponent().groupCount().by('gremlin.connectedComponentVertexProgram.component').select(values).project('components',
'largest').by(count(local)).by(max(local))
# alg: unweighted shortest path src -> sink over out(), as the vertex count of
the shortest path; the recipe is bounded
# at 3 hops and shortestPath() at maxDistance 3
alg-sp-recipe-p12 | small | * |
g.V(3440L).repeat(out().simplePath()).until(or(hasId(2166L),
loops().is(3))).hasId(2166L).path().count(local).min()
-alg-sp-vp-p12 | small | tinkergraph-computer,spark |
g.V(3440L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.target,
hasId(2166L)).with(ShortestPath.maxDistance, 3).count(local).min()
+alg-sp-vp-p12 | small | tinkergraph-computer,spark,csr-computer |
g.V(3440L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.target,
hasId(2166L)).with(ShortestPath.maxDistance, 3).count(local).min()
alg-sp-recipe-p16 | small | * |
g.V(57232L).repeat(out().simplePath()).until(or(hasId(10358L),
loops().is(3))).hasId(10358L).path().count(local).min()
-alg-sp-vp-p16 | small | tinkergraph-computer,spark |
g.V(57232L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.target,
hasId(10358L)).with(ShortestPath.maxDistance, 3).count(local).min()
+alg-sp-vp-p16 | small | tinkergraph-computer,spark,csr-computer |
g.V(57232L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.target,
hasId(10358L)).with(ShortestPath.maxDistance, 3).count(local).min()
alg-sp-recipe-p18 | small | * |
g.V(97549L).repeat(out().simplePath()).until(or(hasId(10358L),
loops().is(3))).hasId(10358L).path().count(local).min()
-alg-sp-vp-p18 | small | tinkergraph-computer,spark |
g.V(97549L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.target,
hasId(10358L)).with(ShortestPath.maxDistance, 3).count(local).min()
+alg-sp-vp-p18 | small | tinkergraph-computer,spark,csr-computer |
g.V(97549L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.target,
hasId(10358L)).with(ShortestPath.maxDistance, 3).count(local).min()
# alg: weighted shortest path wsrc -> sink over out() by the edge weight, as
the vertex and edge ids of the lightest
# path; the recipe is bounded at 3 hops, shortestPath() is not (maxDistance
only bounds hop counts). The distance itself
# cannot be compared: shortestPath() returns paths of reference elements,
whose weight cannot be read back on a graph
# computer. CsrAlgorithmSources prints the distance.
alg-wsp-recipe-p12 | small | * |
g.withSack(0.0d).V(3440L).repeat(outE().sack(sum).by('weight').inV().simplePath()).until(or(hasId(2166L),
loops().is(3))).hasId(2166L).order().by(sack()).limit(1).path().map(unfold().id().fold())
-alg-wsp-vp-p12 | small | tinkergraph-computer,spark |
g.V(3440L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.distance, 'weight').with(ShortestPath.target,
hasId(2166L)).with(ShortestPath.includeEdges,
true).limit(1).map(unfold().id().fold())
+alg-wsp-vp-p12 | small | tinkergraph-computer,spark,csr-computer |
g.V(3440L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.distance, 'weight').with(ShortestPath.target,
hasId(2166L)).with(ShortestPath.includeEdges,
true).limit(1).map(unfold().id().fold())
alg-wsp-recipe-p16 | small | * |
g.withSack(0.0d).V(525L).repeat(outE().sack(sum).by('weight').inV().simplePath()).until(or(hasId(10358L),
loops().is(3))).hasId(10358L).order().by(sack()).limit(1).path().map(unfold().id().fold())
-alg-wsp-vp-p16 | small | tinkergraph-computer,spark |
g.V(525L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.distance, 'weight').with(ShortestPath.target,
hasId(10358L)).with(ShortestPath.includeEdges,
true).limit(1).map(unfold().id().fold())
+alg-wsp-vp-p16 | small | tinkergraph-computer,spark,csr-computer |
g.V(525L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.distance, 'weight').with(ShortestPath.target,
hasId(10358L)).with(ShortestPath.includeEdges,
true).limit(1).map(unfold().id().fold())
alg-wsp-recipe-p18 | small | * |
g.withSack(0.0d).V(33986L).repeat(outE().sack(sum).by('weight').inV().simplePath()).until(or(hasId(10358L),
loops().is(3))).hasId(10358L).order().by(sack()).limit(1).path().map(unfold().id().fold())
-alg-wsp-vp-p18 | small | tinkergraph-computer,spark |
g.V(33986L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.distance, 'weight').with(ShortestPath.target,
hasId(10358L)).with(ShortestPath.includeEdges,
true).limit(1).map(unfold().id().fold())
+alg-wsp-vp-p18 | small | tinkergraph-computer,spark,csr-computer |
g.V(33986L).shortestPath().with(ShortestPath.edges,
Direction.OUT).with(ShortestPath.distance, 'weight').with(ShortestPath.target,
hasId(10358L)).with(ShortestPath.includeEdges,
true).limit(1).map(unfold().id().fold())
# alg: single-source both() distances from bsrc, as the vertex counts at
distance 1..k in distance order. The
# recipe groups by loops() inside the repeat (dedup() keeps the first,
shortest, visit); shortestPath() keeps one path
# per end vertex and groups by path length.
alg-bfs-recipe-p12 | small | * |
g.V(1759L).repeat(both().not(hasId(1759L)).dedup().groupCount('m').by(loops())).times(3).cap('m').order(local).by(keys).select(values)
-alg-bfs-vp-p12 | small | tinkergraph-computer,spark |
g.V(1759L).shortestPath().with(ShortestPath.maxDistance,
3).dedup().by(tail(local,
1)).filter(count(local).is(gt(1))).groupCount().by(count(local)).order(local).by(keys).select(values)
+alg-bfs-vp-p12 | small | tinkergraph-computer,spark,csr-computer |
g.V(1759L).shortestPath().with(ShortestPath.maxDistance,
3).dedup().by(tail(local,
1)).filter(count(local).is(gt(1))).groupCount().by(count(local)).order(local).by(keys).select(values)
alg-bfs-recipe-p16 | small | * |
g.V(1764L).repeat(both().not(hasId(1764L)).dedup().groupCount('m').by(loops())).times(3).cap('m').order(local).by(keys).select(values)
-alg-bfs-vp-p16 | small | tinkergraph-computer,spark |
g.V(1764L).shortestPath().with(ShortestPath.maxDistance,
3).dedup().by(tail(local,
1)).filter(count(local).is(gt(1))).groupCount().by(count(local)).order(local).by(keys).select(values)
+alg-bfs-vp-p16 | small | tinkergraph-computer,spark,csr-computer |
g.V(1764L).shortestPath().with(ShortestPath.maxDistance,
3).dedup().by(tail(local,
1)).filter(count(local).is(gt(1))).groupCount().by(count(local)).order(local).by(keys).select(values)
alg-bfs-recipe-p18 | small | * |
g.V(1660L).repeat(both().not(hasId(1660L)).dedup().groupCount('m').by(loops())).times(3).cap('m').order(local).by(keys).select(values)
-alg-bfs-vp-p18 | small | tinkergraph-computer,spark |
g.V(1660L).shortestPath().with(ShortestPath.maxDistance,
3).dedup().by(tail(local,
1)).filter(count(local).is(gt(1))).groupCount().by(count(local)).order(local).by(keys).select(values)
+alg-bfs-vp-p18 | small | tinkergraph-computer,spark,csr-computer |
g.V(1660L).shortestPath().with(ShortestPath.maxDistance,
3).dedup().by(tail(local,
1)).filter(count(local).is(gt(1))).groupCount().by(count(local)).order(local).by(keys).select(values)
# alg: directed three-cycles, small graph (powerlaw-12) only. The first counts
closed out-walks of length three
# (parallel edges and self-loops included); the second is the cycle-detection
recipe and counts distinct triangles.
