fix mistakes: should be able to call with or without mergeFunc

larryxiao · larryxiao · commit efae765bfb67 · 2014-08-19T10:39:53.000+08:00
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/VertexRDD.scala b/graphx/src/main/scala/org/apache/spark/graphx/VertexRDD.scala
@@ -419,7 +419,7 @@ object VertexRDD {
       (vertexIter, routingTableIter) =>
         val routingTable =
           if (routingTableIter.hasNext) routingTableIter.next() else RoutingTablePartition.empty
-        Iterator(ShippableVertexPartition(vertexIter, routingTable, defaultVal))
+        Iterator(ShippableVertexPartition(vertexIter, routingTable, defaultVal, mergeFunc))
     }
     new VertexRDD(vertexPartitions)
   }
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/impl/ShippableVertexPartition.scala b/graphx/src/main/scala/org/apache/spark/graphx/impl/ShippableVertexPartition.scala
@@ -36,7 +36,16 @@ private[graphx]
 object ShippableVertexPartition {
   /** Construct a `ShippableVertexPartition` from the given vertices without any routing table. */
   def apply[VD: ClassTag](iter: Iterator[(VertexId, VD)]): ShippableVertexPartition[VD] =
-    apply(iter, RoutingTablePartition.empty, null.asInstanceOf[VD])
+    apply(iter, RoutingTablePartition.empty, null.asInstanceOf[VD], (a, b) => a)
+
+  /**
+   * Construct a `ShippableVertexPartition` from the given vertices with the specified routing
+   * table, filling in missing vertices mentioned in the routing table using `defaultVal`.
+   */
+  def apply[VD: ClassTag](
+    iter: Iterator[(VertexId, VD)], routingTable: RoutingTablePartition, defaultVal: VD)
+    : ShippableVertexPartition[VD] = 
+    apply(iter, routingTable, defaultVal, (a, b) => a)
 
   /**
    * Construct a `ShippableVertexPartition` from the given vertices with the specified routing

Original file line number	Diff line number	Diff line change
`@@ -419,7 +419,7 @@ object VertexRDD {`
`419`	`419`	`(vertexIter, routingTableIter) =>`
`420`	`420`	`val routingTable =`
`421`	`421`	`if (routingTableIter.hasNext) routingTableIter.next() else RoutingTablePartition.empty`
`422`		`- Iterator(ShippableVertexPartition(vertexIter, routingTable, defaultVal))`
	`422`	`+ Iterator(ShippableVertexPartition(vertexIter, routingTable, defaultVal, mergeFunc))`
`423`	`423`	`}`
`424`	`424`	`new VertexRDD(vertexPartitions)`
`425`	`425`	`}`