updated to aggregateUsingIndex and added test to ensure that method works properly

Author: Brennon York

graphx/src/main/scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala

@@ -104,7 +104,7 @@ class VertexRDDImpl[VD] private[graphx] (
     this.mapVertexPartitions(_.map(f))
 
   override def diff(other: RDD[(VertexId, VD)]): VertexRDD[VD] = {
-    diff(VertexRDD(other))
+    diff(this.aggregateUsingIndex(other, (a: VD, b: VD) => a))
   }
 
   override def diff(other: VertexRDD[VD]): VertexRDD[VD] = {

graphx/src/test/scala/org/apache/spark/graphx/VertexRDDSuite.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.graphx
 
+import org.apache.spark.rdd.RDD
 import org.scalatest.FunSuite
 
 import org.apache.spark.SparkContext
@@ -58,6 +59,18 @@ class VertexRDDSuite extends FunSuite with LocalSparkContext {
     }
   }
 
+  test("diff with RDD[(VertexId, VD)]") {
+    withSpark { sc =>
+      val n = 100
+      val verts = vertices(sc, n).cache()
+      val flipEvens: RDD[(VertexId, Int)] =
+        sc.parallelize(0L to 100L)
+          .map(id => if (id % 2 == 0) (id, -id.toInt) else (id, id.toInt)).cache()
+      // diff should keep only the changed vertices
+      assert(verts.diff(flipEvens).map(_._2).collect().toSet === (2 to n by 2).map(-_).toSet)
+    }
+  }
+
   test("leftJoin") {
     withSpark { sc =>
       val n = 100