You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ok...@apache.org on 2016/03/31 21:10:58 UTC
[39/40] incubator-tinkerpop git commit: Travis failed -- locally
worked. Insidious bug. Serialization ordering caused a problem -- had to
slightly alter my model. Fortunately,
my alteration is more efficient than the previous push. So BAM.
Travis failed -- locally worked. Insidious bug. Serialization ordering caused a problem -- had to slightly alter my model. Fortunately, my alteration is more efficient than the previous push. So BAM.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/691ea472
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/691ea472
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/691ea472
Branch: refs/heads/TINKERPOP-1225
Commit: 691ea4727050ccbe442a1650bcc2ab9cbe4eb798
Parents: 4b0784c
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed Mar 30 17:22:15 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu Mar 31 12:52:07 2016 -0600
----------------------------------------------------------------------
.../process/traversal/step/map/GroupStep.java | 58 ++++++++++----------
.../step/sideEffect/GroupSideEffectStep.java | 40 +++++++++-----
.../step/util/ReducingBarrierStep.java | 44 +++++----------
3 files changed, 69 insertions(+), 73 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/691ea472/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
index a8cf550..ff179ba 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
@@ -57,8 +57,9 @@ public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>>
private char state = 'k';
private Traversal.Admin<S, K> keyTraversal = null;
- public Traversal.Admin<S, V> valueTraversal = this.integrateChild(__.fold().asAdmin());
- public Traversal.Admin<S, ?> preTraversal = null; // used in OLAP
+ private Traversal.Admin<S, V> valueTraversal = this.integrateChild(__.fold().asAdmin());
+ private Traversal.Admin<S, ?> preTraversal = null; // used in OLAP
+ private ReducingBarrierStep reducingBarrierStep = null; // used in OLAP
private boolean onGraphComputer = false;
public GroupStep(final Traversal.Admin traversal) {
@@ -85,15 +86,20 @@ public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>>
public Map<K, V> projectTraverser(final Traverser.Admin<S> traverser) {
final Map<K, V> map = new HashMap<>(1);
final K key = TraversalUtil.applyNullable(traverser, this.keyTraversal);
- if (null == this.preTraversal)
+ if (this.onGraphComputer) {
+ if (null == this.reducingBarrierStep) {
+ final TraverserSet traverserSet = new TraverserSet();
+ this.preTraversal.reset();
+ this.preTraversal.addStart(traverser.split());
+ this.preTraversal.getEndStep().forEachRemaining(traverserSet::add);
+ map.put(key, (V) traverserSet);
+ } else {
+ this.valueTraversal.reset();
+ this.valueTraversal.addStart(traverser.split());
+ map.put(key, (V) this.reducingBarrierStep.nextBarrier());
+ }
+ } else
map.put(key, (V) traverser);
- else { // OLAP no reducing barrier
- final TraverserSet traverserSet = new TraverserSet();
- this.preTraversal.reset();
- this.preTraversal.addStart(traverser.split());
- this.preTraversal.getEndStep().forEachRemaining(traverserSet::add);
- map.put(key, (V) traverserSet);
- }
return map;
}
@@ -124,6 +130,9 @@ public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>>
clone.valueTraversal = this.valueTraversal.clone();
if (null != this.preTraversal)
clone.preTraversal = this.preTraversal.clone();
+ final Optional<Barrier> optional = TraversalHelper.getFirstStepOfAssignableClass(Barrier.class, clone.valueTraversal);
+ if (optional.isPresent() && optional.get() instanceof ReducingBarrierStep)
+ clone.reducingBarrierStep = (ReducingBarrierStep) optional.get();
return clone;
}
@@ -150,9 +159,10 @@ public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>>
@Override
public void onGraphComputer() {
+ this.preTraversal = this.integrateChild(splitOnBarrierStep(this.valueTraversal).get(0));
final Optional<Barrier> optional = TraversalHelper.getFirstStepOfAssignableClass(Barrier.class, this.valueTraversal);
- if (!(optional.isPresent() && optional.get() instanceof ReducingBarrierStep))
- this.preTraversal = this.integrateChild(splitOnBarrierStep(this.valueTraversal).get(0));
+ if (optional.isPresent() && optional.get() instanceof ReducingBarrierStep)
+ this.reducingBarrierStep = (ReducingBarrierStep) optional.get();
this.setReducingBiOperator(new GroupBiOperator<>(this.valueTraversal, this.onGraphComputer = true));
}
@@ -190,21 +200,14 @@ public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>>
if (this.hasReducingBarrier) {
// if there is a mid-barrier, process using midway reductions (reduces memory footprint)
final Traversal.Admin valueTraversalClone = this.valueTraversal.clone(); // be nice to just reset() but TinkerGraphComputer has thread issues
- final ReducingBarrierStep barrier = TraversalHelper.getFirstStepOfAssignableClass(ReducingBarrierStep.class, valueTraversalClone).get();
+ final ReducingBarrierStep reducingBarrierStep = TraversalHelper.getFirstStepOfAssignableClass(ReducingBarrierStep.class, valueTraversalClone).get();
final Object objectA = mapA.get(key);
- if (objectA instanceof Traverser.Admin) {
- barrier.processAllStarts();
- valueTraversalClone.addStart((Traverser.Admin) objectA);
- } else if (null != objectA)
- barrier.addBarrier(objectA);
- if (objectB instanceof Traverser.Admin) {
- barrier.processAllStarts();
- valueTraversalClone.addStart((Traverser.Admin) objectB);
- } else
- barrier.addBarrier(objectB);
+ if (null != objectA)
+ reducingBarrierStep.addBarrier(objectA);
+ reducingBarrierStep.addBarrier(objectB);
// only store the reduction barrier (i.e. seed), not the traversal
- if (barrier.hasNextBarrier())
- mapA.put(key, (V) barrier.nextBarrier());
+ if (reducingBarrierStep.hasNextBarrier())
+ mapA.put(key, (V) reducingBarrierStep.nextBarrier());
} else {
// if there is no mid-traversal reducer, aggregate pre-barrier traversers into a traverser set (expensive, but that's that)
final Object objectA = mapA.get(key);
@@ -280,10 +283,7 @@ public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>>
if (onGraphComputer) {
if (hasReducingBarrier) { // OLAP with reduction (barrier)
final Traversal.Admin<?, V> valueTraversalClone = valueTraversal.clone();
- if (entry.getValue() instanceof Traverser.Admin)
- valueTraversalClone.addStart((Traverser.Admin) entry.getValue());
- else
- TraversalHelper.getFirstStepOfAssignableClass(Barrier.class, valueTraversalClone).get().addBarrier(entry.getValue());
+ TraversalHelper.getFirstStepOfAssignableClass(Barrier.class, valueTraversalClone).get().addBarrier(entry.getValue());
reducedMap.put(entry.getKey(), valueTraversalClone.next());
} else { // OLAP without reduction (traverser set)
postTraversal.reset();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/691ea472/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java
index 38fe6c1..8ec66c0 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java
@@ -50,7 +50,8 @@ public final class GroupSideEffectStep<S, K, V> extends SideEffectStep<S> implem
private char state = 'k';
private Traversal.Admin<S, K> keyTraversal = null;
private Traversal.Admin<S, V> valueTraversal = this.integrateChild(__.fold().asAdmin());
- public Traversal.Admin<S, ?> preTraversal = null; // used in OLAP
+ private Traversal.Admin<S, ?> preTraversal = null; // used in OLAP
+ private ReducingBarrierStep reducingBarrierStep = null; // used in OLAP
private boolean onGraphComputer = false;
///
private String sideEffectKey;
@@ -78,16 +79,21 @@ public final class GroupSideEffectStep<S, K, V> extends SideEffectStep<S> implem
@Override
protected void sideEffect(final Traverser.Admin<S> traverser) {
final Map<K, Object> map = new HashMap<>(1);
- final K key = TraversalUtil.applyNullable(traverser, keyTraversal);
- if (null == this.preTraversal) {
+ final K key = TraversalUtil.applyNullable(traverser, this.keyTraversal);
+ if (this.onGraphComputer) {
+ if (null == this.reducingBarrierStep) {
+ final TraverserSet traverserSet = new TraverserSet<>();
+ this.preTraversal.reset();
+ this.preTraversal.addStart(traverser.split());
+ this.preTraversal.getEndStep().forEachRemaining(traverserSet::add);
+ map.put(key, traverserSet);
+ } else {
+ this.valueTraversal.reset();
+ this.valueTraversal.addStart(traverser.split());
+ map.put(key, (V) this.reducingBarrierStep.nextBarrier());
+ }
+ } else
map.put(key, traverser.split());
- } else { // OLAP no reducing barrier
- final TraverserSet traverserSet = new TraverserSet<>();
- this.preTraversal.reset();
- this.preTraversal.addStart(traverser.split());
- this.preTraversal.getEndStep().forEachRemaining(traverserSet::add);
- map.put(key, traverserSet);
- }
this.getTraversal().getSideEffects().add(this.sideEffectKey, map);
}
@@ -103,7 +109,7 @@ public final class GroupSideEffectStep<S, K, V> extends SideEffectStep<S> implem
@Override
public List<Traversal.Admin<?, ?>> getLocalChildren() {
- final List<Traversal.Admin<?, ?>> children = new ArrayList<>(4);
+ final List<Traversal.Admin<?, ?>> children = new ArrayList<>(2);
if (null != this.keyTraversal)
children.add((Traversal.Admin) this.keyTraversal);
children.add(this.valueTraversal);
@@ -123,6 +129,9 @@ public final class GroupSideEffectStep<S, K, V> extends SideEffectStep<S> implem
clone.valueTraversal = this.valueTraversal.clone();
if (null != this.preTraversal)
clone.preTraversal = this.preTraversal.clone();
+ final Optional<Barrier> optional = TraversalHelper.getFirstStepOfAssignableClass(Barrier.class, clone.valueTraversal);
+ if (optional.isPresent() && optional.get() instanceof ReducingBarrierStep)
+ clone.reducingBarrierStep = (ReducingBarrierStep) optional.get();
return clone;
}
@@ -143,15 +152,16 @@ public final class GroupSideEffectStep<S, K, V> extends SideEffectStep<S> implem
}
@Override
- public Map<K, V> generateFinalResult(final Map<K, ?> unGroupedMap) {
- return GroupStep.doFinalReduction((Map<K, Object>) unGroupedMap, this.valueTraversal, this.onGraphComputer);
+ public Map<K, V> generateFinalResult(final Map<K, ?> object) {
+ return GroupStep.doFinalReduction((Map<K, Object>) object, this.valueTraversal, this.onGraphComputer);
}
@Override
public void onGraphComputer() {
+ this.preTraversal = this.integrateChild(GroupStep.splitOnBarrierStep(this.valueTraversal).get(0));
final Optional<Barrier> optional = TraversalHelper.getFirstStepOfAssignableClass(Barrier.class, this.valueTraversal);
- if (!(optional.isPresent() && optional.get() instanceof ReducingBarrierStep))
- this.preTraversal = this.integrateChild(GroupStep.splitOnBarrierStep(this.valueTraversal).get(0));
+ if (optional.isPresent() && optional.get() instanceof ReducingBarrierStep)
+ this.reducingBarrierStep = (ReducingBarrierStep) optional.get();
this.getTraversal().getSideEffects().register(this.sideEffectKey, null, new GroupStep.GroupBiOperator<>(this.valueTraversal, this.onGraphComputer = true));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/691ea472/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java
index 5113431..3b2823c 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java
@@ -26,7 +26,6 @@ import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
import org.apache.tinkerpop.gremlin.process.traversal.step.Generating;
import org.apache.tinkerpop.gremlin.process.traversal.util.FastNoSuchElementException;
-import java.util.Iterator;
import java.util.function.BinaryOperator;
import java.util.function.Supplier;
@@ -38,7 +37,7 @@ public abstract class ReducingBarrierStep<S, E> extends AbstractStep<S, E> imple
protected Supplier<E> seedSupplier;
protected BinaryOperator<E> reducingBiOperator;
- private boolean done = false;
+ private boolean hasProcessedOnce = false;
private E seed = null;
public ReducingBarrierStep(final Traversal.Admin traversal) {
@@ -65,31 +64,21 @@ public abstract class ReducingBarrierStep<S, E> extends AbstractStep<S, E> imple
public void reset() {
super.reset();
- this.done = false;
+ this.hasProcessedOnce = false;
this.seed = null;
}
@Override
- public void addStarts(final Iterator<Traverser.Admin<S>> starts) {
- if (starts.hasNext()) {
- this.done = false;
- super.addStarts(starts);
- }
- }
-
- @Override
public void done() {
- this.done = true;
- }
-
- @Override
- public void addStart(final Traverser.Admin<S> start) {
- this.done = false;
- super.addStart(start);
+ this.hasProcessedOnce = true;
+ this.seed = null;
}
@Override
public void processAllStarts() {
+ if (this.hasProcessedOnce && !this.starts.hasNext())
+ return;
+ this.hasProcessedOnce = true;
if (this.seed == null) this.seed = this.seedSupplier.get();
while (this.starts.hasNext())
this.seed = this.reducingBiOperator.apply(this.seed, this.projectTraverser(this.starts.next()));
@@ -98,35 +87,32 @@ public abstract class ReducingBarrierStep<S, E> extends AbstractStep<S, E> imple
@Override
public boolean hasNextBarrier() {
this.processAllStarts();
- return !this.done;
+ return null != this.seed;
}
@Override
public E nextBarrier() {
- if (this.done)
+ if (!this.hasNextBarrier())
throw FastNoSuchElementException.instance();
else {
- processAllStarts();
final E temp = this.seed;
this.seed = null;
- this.done = true;
return temp;
}
}
@Override
public void addBarrier(final E barrier) {
- this.processAllStarts();
- this.done = false;
- this.seed = this.reducingBiOperator.apply(this.seed, barrier);
+ this.seed = null == this.seed ?
+ barrier :
+ this.reducingBiOperator.apply(this.seed, barrier);
}
@Override
public Traverser.Admin<E> processNextStart() {
- if (this.done)
- throw FastNoSuchElementException.instance();
this.processAllStarts();
- this.done = true;
+ if (this.seed == null)
+ throw FastNoSuchElementException.instance();
final Traverser.Admin<E> traverser = this.getTraversal().getTraverserGenerator().generate(this.generateFinalResult(this.seed), (Step<E, E>) this, 1l);
this.seed = null;
return traverser;
@@ -135,7 +121,7 @@ public abstract class ReducingBarrierStep<S, E> extends AbstractStep<S, E> imple
@Override
public ReducingBarrierStep<S, E> clone() {
final ReducingBarrierStep<S, E> clone = (ReducingBarrierStep<S, E>) super.clone();
- clone.done = false;
+ clone.hasProcessedOnce = false;
clone.seed = null;
return clone;
}