Add parameter to RDD.toDebugString to allow detailed memory info to be shown or not. Default is for it not to be shown.

Author: Nathan Kronenfeld

core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala

@@ -487,7 +487,12 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
 
   /** A description of this RDD and its recursive dependencies for debugging. */
   def toDebugString(): String = {
-    rdd.toDebugString
+    rdd.toDebugString()
+  }
+
+  /** A description of this RDD and its recursive dependencies for debugging. */
+  def toDebugString(debugMemory: Boolean): String = {
+    rdd.toDebugString(debugMemory)
   }
 
   /**

core/src/main/scala/org/apache/spark/rdd/RDD.scala

@@ -1268,16 +1268,19 @@ abstract class RDD[T: ClassTag](
   }
 
   /** A description of this RDD and its recursive dependencies for debugging. */
-  def toDebugString: String = {
+  @DeveloperApi
+  def toDebugString (debugMemory: Boolean = false): String = {
     // Get a debug description of an rdd without its children
     def debugSelf (rdd: RDD[_]): Seq[String] = {
       import Utils.bytesToString
 
       val persistence = storageLevel.description
-      val storageInfo = rdd.context.getRDDStorageInfo.filter(_.id == rdd.id).map(info =>
-        "    CachedPartitions: %d; MemorySize: %s; TachyonSize: %s; DiskSize: %s".format(
-          info.numCachedPartitions, bytesToString(info.memSize),
-          bytesToString(info.tachyonSize), bytesToString(info.diskSize)))
+      val storageInfo = if (debugMemory)
+	      rdd.context.getRDDStorageInfo.filter(_.id == rdd.id).map(info =>
+		      "    CachedPartitions: %d; MemorySize: %s; TachyonSize: %s; DiskSize: %s".format(
+			      info.numCachedPartitions, bytesToString(info.memSize),
+			      bytesToString(info.tachyonSize), bytesToString(info.diskSize)))
+      else Array[String]()
 
       s"$rdd [$persistence]" +: storageInfo
     }

core/src/test/scala/org/apache/spark/CheckpointSuite.scala

@@ -255,13 +255,13 @@ class CheckpointSuite extends FunSuite with LocalSparkContext with Logging {
     val partitionsBeforeCheckpoint = operatedRDD.partitions
 
     // Find serialized sizes before and after the checkpoint
-    logInfo("RDD after checkpoint: " + operatedRDD + "\n" + operatedRDD.toDebugString)
+    logInfo("RDD after checkpoint: " + operatedRDD + "\n" + operatedRDD.toDebugString())
     val (rddSizeBeforeCheckpoint, partitionSizeBeforeCheckpoint) = getSerializedSizes(operatedRDD)
     operatedRDD.checkpoint()
     val result = operatedRDD.collect()
     operatedRDD.collect() // force re-initialization of post-checkpoint lazy variables
     val (rddSizeAfterCheckpoint, partitionSizeAfterCheckpoint) = getSerializedSizes(operatedRDD)
-    logInfo("RDD after checkpoint: " + operatedRDD + "\n" + operatedRDD.toDebugString)
+    logInfo("RDD after checkpoint: " + operatedRDD + "\n" + operatedRDD.toDebugString())
 
     // Test whether the checkpoint file has been created
     assert(sc.checkpointFile[U](operatedRDD.getCheckpointFile.get).collect() === result)
@@ -313,13 +313,13 @@ class CheckpointSuite extends FunSuite with LocalSparkContext with Logging {
     initializeRdd(operatedRDD)
 
     // Find serialized sizes before and after the checkpoint
-    logInfo("RDD after checkpoint: " + operatedRDD + "\n" + operatedRDD.toDebugString)
+    logInfo("RDD after checkpoint: " + operatedRDD + "\n" + operatedRDD.toDebugString())
     val (rddSizeBeforeCheckpoint, partitionSizeBeforeCheckpoint) = getSerializedSizes(operatedRDD)
     parentRDDs.foreach(_.checkpoint())  // checkpoint the parent RDD, not the generated one
     val result = operatedRDD.collect()  // force checkpointing
     operatedRDD.collect() // force re-initialization of post-checkpoint lazy variables
     val (rddSizeAfterCheckpoint, partitionSizeAfterCheckpoint) = getSerializedSizes(operatedRDD)
-    logInfo("RDD after checkpoint: " + operatedRDD + "\n" + operatedRDD.toDebugString)
+    logInfo("RDD after checkpoint: " + operatedRDD + "\n" + operatedRDD.toDebugString())
 
     // Test whether the data in the checkpointed RDD is same as original
     assert(operatedRDD.collect() === result)