[FLINK-1514] [gelly] renamed fields in Neighbor class; added forwarde…

…d fields hint in GSAIteration
apache · Apr 25, 2015 · 66d72ac · 66d72ac
1 parent 4e3165e
commit 66d72ac
Show file tree

Hide file tree

Showing 4 changed files with 11 additions and 7 deletions.
diff --git a/...ink-gelly/src/main/java/org/apache/flink/graph/example/GSAConnectedComponentsExample.java b/...ink-gelly/src/main/java/org/apache/flink/graph/example/GSAConnectedComponentsExample.java
@@ -89,7 +89,7 @@ public Long map(Long vertexId) {
 	private static final class GatherNeighborIds extends GatherFunction<Long, NullValue, Long> {
 
 		public Long gather(Neighbor<Long, NullValue> neighbor) {
-			return neighbor.getSrcVertexValue();
+			return neighbor.getNeighborValue();
 		}
 	};
 

diff --git a/...lly/src/main/java/org/apache/flink/graph/example/GSASingleSourceShortestPathsExample.java b/...lly/src/main/java/org/apache/flink/graph/example/GSASingleSourceShortestPathsExample.java
@@ -97,8 +97,8 @@ public Double map(Long id) {
 	@SuppressWarnings("serial")
 	private static final class CalculateDistances extends GatherFunction<Double, Double, Double> {
 
-		public Double gather(Neighbor<Double, Double> richEdge) {
+		public Double gather(Neighbor<Double, Double> neighbor) {
-			return richEdge.getSrcVertexValue() + richEdge.getEdgeValue();
+			return neighbor.getNeighborValue() + neighbor.getEdgeValue();
 		}
 	};
 

diff --git a/...staging/flink-gelly/src/main/java/org/apache/flink/graph/gsa/GatherSumApplyIteration.java b/...staging/flink-gelly/src/main/java/org/apache/flink/graph/gsa/GatherSumApplyIteration.java
@@ -26,6 +26,7 @@
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.operators.CustomUnaryOperation;
 import org.apache.flink.api.java.operators.DeltaIteration;
+import org.apache.flink.api.java.operators.JoinOperator;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
 import org.apache.flink.api.java.typeutils.TupleTypeInfo;
@@ -127,12 +128,15 @@ public DataSet<Vertex<K, VV>> createResult() {
 		// Gather, sum and apply
 		DataSet<Tuple2<K, M>> gatheredSet = neighbors.map(gatherUdf);
 		DataSet<Tuple2<K, M>> summedSet = gatheredSet.groupBy(0).reduce(sumUdf);
-		DataSet<Vertex<K, VV>> appliedSet = summedSet
+		JoinOperator<?, ?, Vertex<K, VV>> appliedSet = summedSet
 				.join(iteration.getSolutionSet())
 				.where(0)
 				.equalTo(0)
 				.with(applyUdf);
 
+		// let the operator know that we preserve the key field
+		appliedSet.withForwardedFieldsFirst("0").withForwardedFieldsSecond("0");
+
 		return iteration.closeWith(appliedSet, appliedSet);
 	}
 

diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/gsa/Neighbor.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/gsa/Neighbor.java
@@ -34,11 +34,11 @@ public class Neighbor<VV extends Serializable, EV extends Serializable>
 
 	public Neighbor() {}
 
-	public Neighbor(VV src, EV edge) {
+	public Neighbor(VV neighborValue, EV edgeValue) {
-		super(src, edge);
+		super(neighborValue, edgeValue);
 	}
 
-	public VV getSrcVertexValue() {
+	public VV getNeighborValue() {
 		return this.f0;
 	}