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 58e761d Really cleaned up Beam. We have a TopologyUtil that
translates a Compilation into a Beam topology. We also have branch support
generalized so union, choose, etc. just work without any special consideration
in Beam. I need to get repeat() to be general -- but that is tricky given its
recrussive nature. For now, a special repeat()-case exists.
58e761d is described below
commit 58e761de63ab5c2f89343ed2a5f70fd7a257ed4f
Author: Marko A. Rodriguez <[email protected]>
AuthorDate: Wed Mar 13 10:09:01 2019 -0600
Really cleaned up Beam. We have a TopologyUtil that translates a
Compilation into a Beam topology. We also have branch support generalized so
union, choose, etc. just work without any special consideration in Beam. I need
to get repeat() to be general -- but that is tricky given its recrussive
nature. For now, a special repeat()-case exists.
---
.../machine/functions/branch/UnionBranch.java | 3 +-
.../machine/traversers/EmptyTraverser.java | 63 +++++++++++
.../org/apache/tinkerpop/machine/beam/Beam.java | 108 ++-----------------
.../apache/tinkerpop/machine/beam/BranchFn.java | 6 +-
.../apache/tinkerpop/machine/beam/RepeatFn.java | 10 +-
.../tinkerpop/machine/beam/util/TopologyUtil.java | 120 +++++++++++++++++++++
.../apache/tinkerpop/machine/beam/BeamTest.java | 13 ++-
7 files changed, 212 insertions(+), 111 deletions(-)
diff --git
a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/branch/UnionBranch.java
b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/branch/UnionBranch.java
index f10e22c..3195a40 100644
---
a/java/core/src/main/java/org/apache/tinkerpop/machine/functions/branch/UnionBranch.java
+++
b/java/core/src/main/java/org/apache/tinkerpop/machine/functions/branch/UnionBranch.java
@@ -52,8 +52,7 @@ public final class UnionBranch<C, S, E> extends
AbstractFunction<C, S, Iterator<
public Iterator<Traverser<C, E>> apply(final Traverser<C, S> traverser) {
final MultiIterator<Traverser<C, E>> iterator = new MultiIterator<>();
for (final Compilation<C, S, E> branch :
this.branches.get(Boolean.TRUE)) {
- branch.getProcessor().addStart(traverser.clone());
- iterator.addIterator(branch.getProcessor());
+ iterator.addIterator(branch.addTraverser(traverser));
}
return iterator;
}
diff --git
a/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/EmptyTraverser.java
b/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/EmptyTraverser.java
new file mode 100644
index 0000000..b2eefb8
--- /dev/null
+++
b/java/core/src/main/java/org/apache/tinkerpop/machine/traversers/EmptyTraverser.java
@@ -0,0 +1,63 @@
+/*
+ * 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.traversers;
+
+import org.apache.tinkerpop.machine.coefficients.Coefficient;
+import org.apache.tinkerpop.machine.functions.CFunction;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class EmptyTraverser<C, S> implements Traverser<C, S> {
+
+ private static final EmptyTraverser INSTANCE = new EmptyTraverser();
+
+ private EmptyTraverser() {
+ // for static instances
+ }
+
+ @Override
+ public Coefficient<C> coefficient() {
+ throw new IllegalStateException(EmptyTraverser.class.getSimpleName() +
" does not contain a coefficient");
+ }
+
+ @Override
+ public S object() {
+ throw new IllegalStateException(EmptyTraverser.class.getSimpleName() +
" does not contain an object");
+ }
+
+ @Override
+ public Path path() {
+ throw new IllegalStateException(EmptyTraverser.class.getSimpleName() +
" does not contain a path");
+ }
+
+ @Override
+ public <E> Traverser<C, E> split(CFunction<C> function, E object) {
+ return INSTANCE;
+ }
+
+ @Override
+ public Traverser<C, S> clone() {
+ return INSTANCE;
+ }
+
+ public static final <C, S> EmptyTraverser<C, S> instance() {
+ return INSTANCE;
+ }
+}
diff --git
a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/Beam.java
b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/Beam.java
index c24f26f..9981082 100644
---
a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/Beam.java
+++
b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/Beam.java
@@ -20,134 +20,44 @@ package org.apache.tinkerpop.machine.beam;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PCollectionList;
-import org.apache.beam.sdk.values.PCollectionTuple;
-import org.apache.beam.sdk.values.TupleTag;
-import org.apache.beam.sdk.values.TupleTagList;
import org.apache.tinkerpop.machine.beam.serialization.TraverserCoder;
+import org.apache.tinkerpop.machine.beam.util.TopologyUtil;
import org.apache.tinkerpop.machine.bytecode.Compilation;
-import org.apache.tinkerpop.machine.coefficients.Coefficient;
-import org.apache.tinkerpop.machine.coefficients.LongCoefficient;
-import org.apache.tinkerpop.machine.functions.BranchFunction;
-import org.apache.tinkerpop.machine.functions.CFunction;
-import org.apache.tinkerpop.machine.functions.FilterFunction;
-import org.apache.tinkerpop.machine.functions.FlatMapFunction;
-import org.apache.tinkerpop.machine.functions.InitialFunction;
-import org.apache.tinkerpop.machine.functions.MapFunction;
-import org.apache.tinkerpop.machine.functions.ReduceFunction;
-import org.apache.tinkerpop.machine.functions.branch.RepeatBranch;
import org.apache.tinkerpop.machine.processor.Processor;
+import org.apache.tinkerpop.machine.traversers.EmptyTraverser;
import org.apache.tinkerpop.machine.traversers.Traverser;
-import org.apache.tinkerpop.machine.traversers.TraverserFactory;
import java.util.ArrayList;
import java.util.Iterator;
-import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
public class Beam<C, S, E> implements Processor<C, S, E> {
+ public static final int MAX_REPETIONS = 10;
+
private final Pipeline pipeline;
public static List<Traverser> OUTPUT = new ArrayList<>(); // FIX THIS!
private final List<Fn> functions = new ArrayList<>();
private Iterator<Traverser<C, E>> iterator = null;
- private final TraverserCoder<C, S> coder = new TraverserCoder<>();
-
public Beam(final Compilation<C, S, E> compilation) {
this.pipeline = Pipeline.create();
- PCollection<Traverser<C, S>> collection =
this.pipeline.apply(Create.of(compilation.getTraverserFactory().create((Coefficient)
LongCoefficient.create(), 1L)));
- collection.setCoder(this.coder);
- for (final CFunction<C> function : compilation.getFunctions()) {
- collection = processFunction(collection,
compilation.getTraverserFactory(), function, false);
- }
- collection.apply(ParDo.of(new OutputStep<>()));
this.pipeline.getOptions().setRunner(new
PipelineOptions.DirectRunner().create(this.pipeline.getOptions()));
- }
-
- private PCollection<Traverser<C, S>> processFunction(
- PCollection<Traverser<C, S>> collection,
- final TraverserFactory<C> traverserFactory,
- final CFunction<?> function,
- final boolean branching) {
- DoFn<Traverser<C, S>, Traverser<C, E>> fn = null;
- if (function instanceof RepeatBranch) {
- final List<PCollection> outputs = new ArrayList<>();
- final TupleTag repeatDone = new TupleTag<>();
- final TupleTag repeatLoop = new TupleTag<>();
- for (int i = 0; i < 10; i++) {
- fn = new RepeatFn((RepeatBranch) function, repeatDone,
repeatLoop);
- PCollectionTuple branches = (PCollectionTuple)
collection.apply(ParDo.of(fn).withOutputTags(repeatLoop,
TupleTagList.of(repeatDone)));
- branches.get(repeatLoop).setCoder(new TraverserCoder());
- branches.get(repeatDone).setCoder(new TraverserCoder());
- outputs.add(branches.get(repeatDone));
- for (final CFunction<C> repeatFunction : ((RepeatBranch<C, S>)
function).getRepeat().getFunctions()) {
- collection =
this.processFunction(branches.get(repeatLoop), traverserFactory,
repeatFunction, true);
- }
- }
- this.functions.add((Fn) fn);
- collection = (PCollection) PCollectionList.of((Iterable)
outputs).apply(Flatten.pCollections());
- collection.setCoder(new TraverserCoder());
- } else if (function instanceof BranchFunction) {
- final BranchFunction<C, Object, Object, Object> branchFunction =
(BranchFunction) function;
- final Map<Object, TupleTag> branches = new LinkedHashMap<>();
- for (final Map.Entry<Object, List<Compilation<C, Object, Object>>>
b : branchFunction.getBranches().entrySet()) {
- branches.put(b.getKey(), new TupleTag());
- }
- final BranchFn<C, Object, Object, Object> branchFn = new
BranchFn<>(branchFunction, branches);
- final List<TupleTag<?>> tags = new ArrayList(branches.values());
- PCollectionTuple collectionTuple = (PCollectionTuple)
collection.apply(ParDo.of((DoFn) branchFn).withOutputTags(tags.get(0),
TupleTagList.of(tags.subList(1, tags.size()))));
- collectionTuple.getAll().values().forEach(c -> c.setCoder(new
TraverserCoder()));
- final List<PCollection<Traverser<C, S>>> collections = new
ArrayList<>();
- for (final Map.Entry<Object, List<Compilation<C, Object, Object>>>
b : branchFunction.getBranches().entrySet()) {
- for (final Compilation<C, Object, Object> compilation :
b.getValue()) {
- PCollection<Traverser<C, S>> branchCollection =
collectionTuple.get(branches.get(b.getKey()));
- for (final CFunction<C> f : compilation.getFunctions()) {
- branchCollection =
this.processFunction(branchCollection, traverserFactory, f, true);
- }
-
- collections.add(branchCollection);
- }
- }
- collection =
PCollectionList.of(collections).apply(Flatten.pCollections());
- this.functions.add(branchFn);
- } else if (function instanceof InitialFunction) {
- fn = new InitialFn((InitialFunction<C, S>) function,
traverserFactory);
- } else if (function instanceof FilterFunction) {
- fn = new FilterFn((FilterFunction<C, S>) function);
- } else if (function instanceof FlatMapFunction) {
- fn = new FlatMapFn<>((FlatMapFunction<C, S, E>) function);
- } else if (function instanceof MapFunction) {
- fn = new MapFn<>((MapFunction<C, S, E>) function);
- } else if (function instanceof ReduceFunction) {
- final ReduceFn<C, S, E> combine = new
ReduceFn<>((ReduceFunction<C, S, E>) function, traverserFactory);
- collection = (PCollection<Traverser<C, S>>)
collection.apply(Combine.globally((ReduceFn) combine));
- this.functions.add(combine);
- } else
- throw new RuntimeException("You need a new step type:" + function);
-
- if (!(function instanceof ReduceFunction) && !(function instanceof
BranchFunction)) {
- if (!branching)
- this.functions.add((Fn) fn);
- collection = (PCollection<Traverser<C, S>>)
collection.apply(ParDo.of((DoFn) fn));
- }
- collection.setCoder(this.coder);
- return collection;
+ final PCollection<Traverser<C, S>> source =
this.pipeline.apply(Create.of(EmptyTraverser.instance()));
+ source.setCoder(new TraverserCoder<>());
+ final PCollection<Traverser<C, E>> sink = TopologyUtil.compile(source,
compilation);
+ sink.apply(ParDo.of(new OutputStep<>()));
}
@Override
public void addStart(final Traverser<C, S> traverser) {
- this.functions.get(0).addStart(traverser);
+ this.functions.get(0).addStart(traverser); // TODO: use side-inputs
}
@Override
diff --git
a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/BranchFn.java
b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/BranchFn.java
index 2f55b5e..f5560d3 100644
---
a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/BranchFn.java
+++
b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/BranchFn.java
@@ -29,12 +29,12 @@ import java.util.Optional;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public class BranchFn<C, S, E, M> extends AbstractFn<C, S, E> {
+public class BranchFn<C, S, E, M> extends AbstractFn<C, S, S> {
- private final Map<M, TupleTag> branches;
+ private final Map<M, TupleTag<Traverser<C, S>>> branches;
private final Selector<C, S, M> branchSelector;
- public BranchFn(final BranchFunction<C, S, E, M> branchFunction, final
Map<M, TupleTag> branches) {
+ public BranchFn(final BranchFunction<C, S, E, M> branchFunction, final
Map<M, TupleTag<Traverser<C, S>>> branches) {
super(branchFunction);
this.branches = branches;
this.branchSelector = branchFunction.getBranchSelector();
diff --git
a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/RepeatFn.java
b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/RepeatFn.java
index 8edb581..1f6b006 100644
---
a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/RepeatFn.java
+++
b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/RepeatFn.java
@@ -32,20 +32,24 @@ public class RepeatFn<C, S> extends AbstractFn<C, S, S> {
private final Compilation<C, S, ?> until;
private final TupleTag repeatDone;
private final TupleTag repeatLoop;
+ private final boolean deadEnd;
- public RepeatFn(final RepeatBranch<C, S> repeatBranch, final TupleTag
repeatDone, final TupleTag repeatLoop) {
+ public RepeatFn(final RepeatBranch<C, S> repeatBranch, final TupleTag
repeatDone, final TupleTag repeatLoop, final boolean deadEnd) {
super(repeatBranch);
this.until = repeatBranch.getUntil();
this.repeatDone = repeatDone;
this.repeatLoop = repeatLoop;
+ this.deadEnd = deadEnd;
}
@ProcessElement
public void processElement(final @DoFn.Element Traverser<C, S> traverser,
final MultiOutputReceiver out) {
- if (this.until.filterTraverser(traverser))
+ if (this.until.filterTraverser(traverser.clone()))
out.get(this.repeatDone).output(traverser.clone());
- else
+ else if (!this.deadEnd)
out.get(this.repeatLoop).output(traverser.clone());
+ else
+ throw new IllegalStateException("There are not enough repetition
to account for this traveral");
}
}
diff --git
a/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/util/TopologyUtil.java
b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/util/TopologyUtil.java
new file mode 100644
index 0000000..933a853
--- /dev/null
+++
b/java/machine/beam/src/main/java/org/apache/tinkerpop/machine/beam/util/TopologyUtil.java
@@ -0,0 +1,120 @@
+/*
+ * 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.beam.util;
+
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionList;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.tinkerpop.machine.beam.Beam;
+import org.apache.tinkerpop.machine.beam.BranchFn;
+import org.apache.tinkerpop.machine.beam.FilterFn;
+import org.apache.tinkerpop.machine.beam.FlatMapFn;
+import org.apache.tinkerpop.machine.beam.InitialFn;
+import org.apache.tinkerpop.machine.beam.MapFn;
+import org.apache.tinkerpop.machine.beam.ReduceFn;
+import org.apache.tinkerpop.machine.beam.RepeatFn;
+import org.apache.tinkerpop.machine.beam.serialization.TraverserCoder;
+import org.apache.tinkerpop.machine.bytecode.Compilation;
+import org.apache.tinkerpop.machine.functions.BranchFunction;
+import org.apache.tinkerpop.machine.functions.CFunction;
+import org.apache.tinkerpop.machine.functions.FilterFunction;
+import org.apache.tinkerpop.machine.functions.FlatMapFunction;
+import org.apache.tinkerpop.machine.functions.InitialFunction;
+import org.apache.tinkerpop.machine.functions.MapFunction;
+import org.apache.tinkerpop.machine.functions.ReduceFunction;
+import org.apache.tinkerpop.machine.functions.branch.RepeatBranch;
+import org.apache.tinkerpop.machine.traversers.Traverser;
+import org.apache.tinkerpop.machine.traversers.TraverserFactory;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class TopologyUtil {
+
+ public static <C, S, E> PCollection<Traverser<C, E>> compile(final
PCollection<Traverser<C, S>> source, final Compilation<C, S, E> compilation) {
+ final TraverserFactory<C> traverserFactory =
compilation.getTraverserFactory();
+ PCollection<Traverser<C, E>> sink = (PCollection) source;
+ for (final CFunction<C> function : compilation.getFunctions()) {
+ sink = TopologyUtil.extend(sink, function, traverserFactory);
+ }
+ return sink;
+ }
+
+ private static <C, S, E, M> PCollection<Traverser<C, E>> extend(final
PCollection<Traverser<C, S>> source, final CFunction<C> function, final
TraverserFactory<C> traverserFactory) {
+ PCollection sink;
+ if (function instanceof MapFunction) {
+ sink = source.apply(ParDo.of(new MapFn<>((MapFunction<C, S, E>)
function)));
+ } else if (function instanceof FilterFunction) {
+ sink = source.apply(ParDo.of(new FilterFn<>((FilterFunction<C, S>)
function)));
+ } else if (function instanceof FlatMapFunction) {
+ sink = source.apply(ParDo.of(new FlatMapFn<>((FlatMapFunction<C,
S, E>) function)));
+ } else if (function instanceof InitialFunction) {
+ sink = source.apply(ParDo.of(new InitialFn<>((InitialFunction<C,
S>) function, traverserFactory)));
+ } else if (function instanceof ReduceFunction) {
+ sink = source.apply(Combine.globally(new
ReduceFn<>((ReduceFunction<C, S, E>) function, traverserFactory)));
+ } else if (function instanceof RepeatBranch) {
+ final RepeatBranch<C, S> repeatFunction = (RepeatBranch<C, S>)
function;
+ final List<PCollection<Traverser<C, S>>> repeatSinks = new
ArrayList<>();
+ final TupleTag<Traverser<C, S>> repeatDone = new TupleTag<>();
+ final TupleTag<Traverser<C, S>> repeatLoop = new TupleTag<>();
+ sink = source;
+ for (int i = 0; i < Beam.MAX_REPETIONS; i++) {
+ final RepeatFn<C, S> fn = new RepeatFn<>(repeatFunction,
repeatDone, repeatLoop, i == Beam.MAX_REPETIONS - 1);
+ final PCollectionTuple outputs = (PCollectionTuple)
sink.apply(ParDo.of(fn).withOutputTags(repeatLoop,
TupleTagList.of(repeatDone)));
+ outputs.getAll().values().forEach(c -> c.setCoder(new
TraverserCoder()));
+ repeatSinks.add(outputs.get(repeatDone));
+ for (final CFunction<C> ff :
repeatFunction.getRepeat().getFunctions()) {
+ sink = TopologyUtil.extend(outputs.get(repeatLoop), ff,
traverserFactory);
+ }
+ }
+ sink =
PCollectionList.of(repeatSinks).apply(Flatten.pCollections());
+ } else if (function instanceof BranchFunction) {
+ final BranchFunction<C, S, E, M> branchFunction =
(BranchFunction<C, S, E, M>) function;
+ final Map<M, TupleTag<Traverser<C, S>>> selectors = new
LinkedHashMap<>();
+ for (final Map.Entry<M, List<Compilation<C, S, E>>> branch :
branchFunction.getBranches().entrySet()) {
+ selectors.put(branch.getKey(), new TupleTag<>());
+ }
+ final BranchFn<C, S, E, M> fn = new BranchFn<>(branchFunction,
selectors);
+ final List<TupleTag<Traverser<C, S>>> tags = new
ArrayList<>(selectors.values());
+ final PCollectionTuple outputs =
source.apply(ParDo.of(fn).withOutputTags(tags.get(0), TupleTagList.of((List)
tags.subList(1, tags.size()))));
+ outputs.getAll().values().forEach(c -> c.setCoder(new
TraverserCoder()));
+ final List<PCollection<Traverser<C, E>>> branchSinks = new
ArrayList<>();
+ for (final Map.Entry<M, List<Compilation<C, S, E>>> branch :
branchFunction.getBranches().entrySet()) {
+ final PCollection<Traverser<C, S>> output =
outputs.get(selectors.get(branch.getKey()));
+ for (final Compilation<C, S, E> compilation :
branch.getValue()) {
+ branchSinks.add(TopologyUtil.compile(output, compilation));
+ }
+ }
+ sink =
PCollectionList.of(branchSinks).apply(Flatten.pCollections());
+ } else
+ throw new RuntimeException("You need a new step type:" + function);
+ sink.setCoder(new TraverserCoder<>());
+ return sink;
+ }
+}
diff --git
a/java/machine/beam/src/test/java/org/apache/tinkerpop/machine/beam/BeamTest.java
b/java/machine/beam/src/test/java/org/apache/tinkerpop/machine/beam/BeamTest.java
index 510b451..b161995 100644
---
a/java/machine/beam/src/test/java/org/apache/tinkerpop/machine/beam/BeamTest.java
+++
b/java/machine/beam/src/test/java/org/apache/tinkerpop/machine/beam/BeamTest.java
@@ -28,7 +28,6 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
-import static org.apache.tinkerpop.language.__.incr;
import static org.apache.tinkerpop.language.__.is;
/**
@@ -42,18 +41,24 @@ public class BeamTest {
.withProcessor(BeamProcessor.class)
.withStrategy(IdentityStrategy.class);
- Traversal<Long, ?, ?> traversal = g.inject(Arrays.asList(2L, 5L,
10L)).<Long>unfold().repeat(__.<Long>incr().identity().map(__.incr())).until(is(10L));
+ Traversal<Long, ?, ?> traversal = g.inject(Arrays.asList(2L, 5L, 3L,
10L)).<Long>unfold().repeat(__.<Long>incr().identity().map(__.incr())).until(is(10L));
System.out.println(TraversalUtil.getBytecode(traversal).getSourceInstructions());
System.out.println(TraversalUtil.getBytecode(traversal));
System.out.println(traversal);
System.out.println(traversal.toList());
System.out.println("\n----------\n");
- traversal =
g.inject(10L).choose(__.is(7L),__.incr(),__.<Long>incr().incr());
+ traversal = g.inject(1L, 2L,
3L).repeat(__.<Long>incr().incr().incr()).until(is(10L));
+
System.out.println(TraversalUtil.getBytecode(traversal).getSourceInstructions());
+ System.out.println(TraversalUtil.getBytecode(traversal));
+ System.out.println(traversal);
+ System.out.println(traversal.toList());
+ System.out.println("\n----------\n");
+ traversal = g.inject(10L).choose(__.is(7L), __.incr(),
__.<Long>incr().incr());
System.out.println(TraversalUtil.getBytecode(traversal));
System.out.println(traversal);
System.out.println(traversal.toList());
System.out.println("\n----------\n");
- traversal =
g.inject(7L).union(__.incr(),__.<Long>incr().incr().union(__.incr(),__.incr()));
+ traversal = g.inject(7L).union(__.incr(),
__.<Long>incr().incr().union(__.incr(), __.incr()));
System.out.println(TraversalUtil.getBytecode(traversal));
System.out.println(traversal);
System.out.println(traversal.toList());
