tinkerpop-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ok...@apache.org
Subject [tinkerpop] branch tp4 updated: 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.
Date Wed, 13 Mar 2019 16:09:45 GMT
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 <okrammarko@gmail.com>
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());


Mime
View raw message