Use abstract class instead of trait for binary compatibility

Author: ankurdave

graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala

@@ -19,6 +19,8 @@ package org.apache.spark.graphx
 
 import scala.reflect.ClassTag
 
+import org.apache.spark.Dependency
+import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 import org.apache.spark.storage.StorageLevel
 
@@ -32,7 +34,9 @@ import org.apache.spark.graphx.impl.EdgeRDDImpl
  * edge to provide the triplet view. Shipping of the vertex attributes is managed by
  * `impl.ReplicatedVertexView`.
  */
-trait EdgeRDD[@specialized ED, VD] extends RDD[Edge[ED]] {
+abstract class EdgeRDD[@specialized ED, VD](
+    @transient sc: SparkContext,
+    @transient deps: Seq[Dependency[_]]) extends RDD[Edge[ED]](sc, deps) {
   /**
    * Map the values in an edge partitioning preserving the structure but changing the values.
    *

graphx/src/main/scala/org/apache/spark/graphx/VertexRDD.scala

@@ -56,7 +56,9 @@ import org.apache.spark.graphx.impl.VertexRDDFunctions._
  *
  * @tparam VD the vertex attribute associated with each vertex in the set.
  */
-trait VertexRDD[@specialized VD] extends RDD[(VertexId, VD)] {
+abstract class VertexRDD[@specialized VD](
+    @transient sc: SparkContext,
+    @transient deps: Seq[Dependency[_]]) extends RDD[(VertexId, VD)](sc, deps) {
 
   implicit protected def vdTag: ClassTag[VD]
 
@@ -81,8 +83,8 @@ trait VertexRDD[@specialized VD] extends RDD[(VertexId, VD)] {
    * preserves the index for efficient joins with the original RDD, and it sets bits in the bitmask
    * rather than allocating new memory.
    *
-   * It is declared and defined in the VertexRDD trait to allow refining the return type from
-   * `RDD[(VertexId, VD)]` to `VertexRDD[VD]`.
+   * It is declared and defined here to allow refining the return type from `RDD[(VertexId, VD)]` to
+   * `VertexRDD[VD]`.
    *
    * @param pred the user defined predicate, which takes a tuple to conform to the
    * `RDD[(VertexId, VD)]` interface

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

@@ -28,8 +28,7 @@ import org.apache.spark.graphx._
 class EdgeRDDImpl[@specialized ED: ClassTag, VD: ClassTag] private[graphx] (
     override val partitionsRDD: RDD[(PartitionID, EdgePartition[ED, VD])],
     val targetStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY)
-  extends RDD[Edge[ED]](partitionsRDD.context, List(new OneToOneDependency(partitionsRDD)))
-  with EdgeRDD[ED, VD] {
+  extends EdgeRDD[ED, VD](partitionsRDD.context, List(new OneToOneDependency(partitionsRDD))) {
 
   override def setName(_name: String): this.type = {
     if (partitionsRDD.name != null) {

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

@@ -32,8 +32,7 @@ class VertexRDDImpl[@specialized VD] private[graphx] (
     val partitionsRDD: RDD[ShippableVertexPartition[VD]],
     val targetStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY)
   (implicit override protected val vdTag: ClassTag[VD])
-  extends RDD[(VertexId, VD)](partitionsRDD.context, List(new OneToOneDependency(partitionsRDD)))
-  with VertexRDD[VD] {
+  extends VertexRDD[VD](partitionsRDD.context, List(new OneToOneDependency(partitionsRDD))) {
 
   require(partitionsRDD.partitioner.isDefined)