This is an automated email from the ASF dual-hosted git repository.
okram pushed a commit to branch tp4
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/tp4 by this push:
new 4cae62b fixed a bug in FilterStep. Added repeat(). I have only tested
it in Pipes. Given that Beam doesn't support recurssion, we will have to use
the exponential loop form developed in the Stream Ring Theory paper. Basically,
concatenate the repeat sequence over and over and over (for some large number)
and always have a multi-output for breaking out of the loop. Need to take a
break.
4cae62b is described below
commit 4cae62b7310a8ad9d1443a11ac6dc502bfd5eda5
Author: Marko A. Rodriguez <[email protected]>
AuthorDate: Tue Mar 12 12:15:53 2019 -0600
fixed a bug in FilterStep. Added repeat(). I have only tested it in Pipes.
Given that Beam doesn't support recurssion, we will have to use the exponential
loop form developed in the Stream Ring Theory paper. Basically, concatenate the
repeat sequence over and over and over (for some large number) and always have
a multi-output for breaking out of the loop. Need to take a break.
---
.../org/apache/tinkerpop/language/Symbols.java | 46 ++++++++++++--
.../org/apache/tinkerpop/language/Traversal.java | 10 +++
.../java/org/apache/tinkerpop/language/__.java | 4 ++
.../tinkerpop/machine/bytecode/BytecodeUtil.java | 3 +
.../tinkerpop/machine/bytecode/Compilation.java | 5 ++
.../machine/functions/branch/RepeatBranch.java | 72 ++++++++++++++++++++++
.../apache/tinkerpop/machine/pipes/FilterStep.java | 10 ++-
.../apache/tinkerpop/machine/pipes/PipesTest.java | 5 +-
8 files changed, 147 insertions(+), 8 deletions(-)
diff --git a/java/core/src/main/java/org/apache/tinkerpop/language/Symbols.java
b/java/core/src/main/java/org/apache/tinkerpop/language/Symbols.java
index 0d431ae..8b4ebef 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/language/Symbols.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/language/Symbols.java
@@ -23,6 +23,10 @@ package org.apache.tinkerpop.language;
*/
public final class Symbols {
+ public static enum Type {
+ INITIAL, MAP, FLATMAP, FILTER, REDUCE, BRANCH
+ }
+
// SOURCE OPS
public static final String WITH_COEFFICIENT = "withCoefficient";
public static final String WITH_PROCESSOR = "withProcessor";
@@ -30,19 +34,53 @@ public final class Symbols {
// INSTRUCTION OPS
- public static final String AS = "as";
- public static final String C = "c";
public static final String COUNT = "count";
public static final String FILTER = "filter";
public static final String GROUP_COUNT = "groupCount";
- public static final String HAS_KEY_VALUE = "hasKeyValue";
public static final String HAS_KEY = "hasKey";
+ public static final String HAS_KEY_VALUE = "hasKeyValue";
public static final String IDENTITY = "identity";
- public static final String IS = "is";
public static final String INCR = "incr";
public static final String INJECT = "inject";
+ public static final String IS = "is";
public static final String MAP = "map";
public static final String PATH = "path";
+ public static final String REPEAT = "repeat";
public static final String SUM = "sum";
public static final String UNION = "union";
+
+ public Type getOpType(final String op) {
+ switch (op) {
+ case COUNT:
+ return Type.REDUCE;
+ case FILTER:
+ return Type.FILTER;
+ case GROUP_COUNT:
+ return Type.REDUCE;
+ case HAS_KEY:
+ return Type.FILTER;
+ case HAS_KEY_VALUE:
+ return Type.FILTER;
+ case IDENTITY:
+ return Type.FILTER;
+ case INCR:
+ return Type.MAP;
+ case INJECT:
+ return Type.INITIAL;
+ case IS:
+ return Type.FILTER;
+ case MAP:
+ return Type.MAP;
+ case PATH:
+ return Type.MAP;
+ case REPEAT:
+ return Type.BRANCH;
+ case SUM:
+ return Type.REDUCE;
+ case UNION:
+ return Type.BRANCH;
+ default:
+ throw new IllegalArgumentException("The following op is
unknown: " + op);
+ }
+ }
}
diff --git
a/java/core/src/main/java/org/apache/tinkerpop/language/Traversal.java
b/java/core/src/main/java/org/apache/tinkerpop/language/Traversal.java
index 92a7d23..cf90d24 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/language/Traversal.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/language/Traversal.java
@@ -143,6 +143,11 @@ public class Traversal<C, S, E> implements Iterator<E> {
return (Traversal) this;
}
+ public Traversal<C, S, E> repeat(final Traversal<C, E, E> repeatTraversal)
{
+ this.bytecode.addInstruction(this.currentCoefficient, Symbols.REPEAT,
repeatTraversal);
+ return this;
+ }
+
public <R extends Number> Traversal<C, S, R> sum() {
this.bytecode.addInstruction(this.currentCoefficient, Symbols.SUM);
return (Traversal) this;
@@ -153,6 +158,11 @@ public class Traversal<C, S, E> implements Iterator<E> {
return (Traversal) this;
}
+ public Traversal<C, S, E> until(final Traversal<C, E, ?> untilTraversal) {
+ this.bytecode.lastInstruction().addArg(untilTraversal);
+ return this;
+ }
+
///////
private final void prepareTraversal() {
diff --git a/java/core/src/main/java/org/apache/tinkerpop/language/__.java
b/java/core/src/main/java/org/apache/tinkerpop/language/__.java
index 8c8a929..d0febb8 100644
--- a/java/core/src/main/java/org/apache/tinkerpop/language/__.java
+++ b/java/core/src/main/java/org/apache/tinkerpop/language/__.java
@@ -41,6 +41,10 @@ public class __ {
return __.<C, Long>start().incr();
}
+ public static <C, S> Traversal<C, S, S> is(final S object) {
+ return __.<C, S>start().is(object);
+ }
+
public static <C, S extends Number> Traversal<C, S, S> sum() {
return __.<C, S>start().sum();
}
diff --git
a/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/BytecodeUtil.java
b/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/BytecodeUtil.java
index 867f112..42922b7 100644
---
a/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/BytecodeUtil.java
+++
b/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/BytecodeUtil.java
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.machine.bytecode;
import org.apache.tinkerpop.language.Symbols;
import org.apache.tinkerpop.machine.coefficients.Coefficient;
import org.apache.tinkerpop.machine.functions.CFunction;
+import org.apache.tinkerpop.machine.functions.branch.RepeatBranch;
import org.apache.tinkerpop.machine.functions.branch.UnionBranch;
import org.apache.tinkerpop.machine.functions.filter.FilterFilter;
import org.apache.tinkerpop.machine.functions.filter.HasKeyFilter;
@@ -156,6 +157,8 @@ public final class BytecodeUtil {
return new MapMap<>(coefficient, labels,
Compilation.compileOne(instruction.args()[0]));
case Symbols.PATH:
return new PathMap<>(coefficient, labels,
Compilation.compile(instruction.args()));
+ case Symbols.REPEAT:
+ return new RepeatBranch(coefficient, labels,
Compilation.compile(instruction.args()));
case Symbols.SUM:
return new SumReduce<>(coefficient, labels);
case Symbols.UNION:
diff --git
a/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/Compilation.java
b/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/Compilation.java
index f973a8f..a3d9b6c 100644
---
a/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/Compilation.java
+++
b/java/core/src/main/java/org/apache/tinkerpop/machine/bytecode/Compilation.java
@@ -55,6 +55,11 @@ public final class Compilation<C, S, E> implements
Serializable {
return this.processor;
}
+ public void addTraverser(final Traverser<C, S> traverser) {
+ this.prepareProcessor();
+ this.processor.addStart(traverser);
+ }
+
public void reset() {
if (null != this.processor)
this.processor.reset();
diff --git
a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/branch/RepeatBranch.java
b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/branch/RepeatBranch.java
new file mode 100644
index 0000000..c879373
--- /dev/null
+++
b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/branch/RepeatBranch.java
@@ -0,0 +1,72 @@
+/*
+ * 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.machine.functions.branch;
+
+import org.apache.tinkerpop.machine.bytecode.Compilation;
+import org.apache.tinkerpop.machine.coefficients.Coefficient;
+import org.apache.tinkerpop.machine.functions.AbstractFunction;
+import org.apache.tinkerpop.machine.functions.BranchFunction;
+import org.apache.tinkerpop.machine.traversers.Traverser;
+import org.apache.tinkerpop.util.StringFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class RepeatBranch<C, S> extends AbstractFunction<C, S,
Iterator<Traverser<C, S>>> implements BranchFunction<C, S, S> {
+
+ private final Compilation<C, S, S> repeat;
+ private final Compilation<C, S, ?> until;
+
+ public RepeatBranch(final Coefficient<C> coefficient, final Set<String>
labels, final List<Compilation<C, S, ?>> repeatCompilations) {
+ super(coefficient, labels);
+ this.repeat = (Compilation<C, S, S>) repeatCompilations.get(0);
+ this.until = repeatCompilations.get(1);
+ }
+
+ @Override
+ public Iterator<Traverser<C, S>> apply(final Traverser<C, S> traverser) {
+ this.repeat.addTraverser(traverser);
+ final List<Traverser<C, S>> toGo = new ArrayList<>();
+ while (this.repeat.getProcessor().hasNext()) {
+ Traverser<C, S> t = this.repeat.getProcessor().next();
+ boolean loop = !this.until.filterTraverser(t);
+ if (loop)
+ this.repeat.getProcessor().addStart(t);
+ else
+ toGo.add(t);
+ }
+ return toGo.iterator();
+ }
+
+ @Override
+ public String toString() {
+ return StringFactory.makeFunctionString(this, this.repeat, this.until);
+ }
+
+ @Override
+ public List<Compilation<C, ?, ?>> getInternals() {
+ return Arrays.asList(this.repeat, this.until);
+ }
+}
diff --git
a/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/FilterStep.java
b/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/FilterStep.java
index 3541017..aebe7db 100644
---
a/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/FilterStep.java
+++
b/java/machine/pipes/src/main/java/org/apache/tinkerpop/machine/pipes/FilterStep.java
@@ -43,7 +43,7 @@ public final class FilterStep<C, S> extends AbstractStep<C,
S, S> {
} else {
Traverser<C, S> traverser;
while (true) {
- traverser = this.getPreviousTraverser();
+ traverser = super.getPreviousTraverser();
if (traverser.filter(this.filterFunction))
return traverser;
}
@@ -57,7 +57,7 @@ public final class FilterStep<C, S> extends AbstractStep<C,
S, S> {
else {
Traverser<C, S> traverser;
while (super.hasNext()) {
- traverser = this.getPreviousTraverser();
+ traverser = super.getPreviousTraverser();
if (traverser.filter(this.filterFunction)) {
this.nextTraverser = traverser;
return true;
@@ -66,4 +66,10 @@ public final class FilterStep<C, S> extends AbstractStep<C,
S, S> {
return false;
}
}
+
+ @Override
+ public void reset() {
+ super.reset();
+ this.nextTraverser = null;
+ }
}
diff --git
a/java/machine/pipes/src/test/java/org/apache/tinkerpop/machine/pipes/PipesTest.java
b/java/machine/pipes/src/test/java/org/apache/tinkerpop/machine/pipes/PipesTest.java
index 691ae6e..9302e5d 100644
---
a/java/machine/pipes/src/test/java/org/apache/tinkerpop/machine/pipes/PipesTest.java
+++
b/java/machine/pipes/src/test/java/org/apache/tinkerpop/machine/pipes/PipesTest.java
@@ -27,7 +27,8 @@ import
org.apache.tinkerpop.machine.coefficients.LongCoefficient;
import org.apache.tinkerpop.machine.strategies.IdentityStrategy;
import org.junit.jupiter.api.Test;
-import java.util.Map;
+import static org.apache.tinkerpop.language.__.incr;
+import static org.apache.tinkerpop.language.__.is;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -41,7 +42,7 @@ public class PipesTest {
.withProcessor(PipesProcessor.class)
.withStrategy(IdentityStrategy.class);
- Traversal<Long, ?, ?> traversal =
g.inject(1L,1L,1L,2L).groupCount().has(__.count(),__.count().incr().incr());
+ Traversal<Long, ?, ?> traversal =
g.inject(1L,1L).repeat(incr()).until(is(10L));
System.out.println(TraversalUtil.getBytecode(traversal));
System.out.println(traversal);
System.out.println(traversal.toList());
