improve and fix compile error

scwf · scwf · commit cdab8854466f · 2016-12-30T15:36:23.000+08:00
diff --git a/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala b/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
@@ -107,7 +107,7 @@ private[spark] class TorrentBroadcast[T: ClassTag](
   def getNumBlocks: Int = numBlocks
 
   /** Total number of blocks this broadcast variable contains. */
-  private val numBlocks: Int = if (!isExecutorSide) writeBlocks(obj) else nBlocks.getOrElse(-1)
+  private val numBlocks: Int = nBlocks.getOrElse(writeBlocks(obj))
 
   /** Whether to generate checksum for blocks or not. */
   private var checksumEnabled: Boolean = false
@@ -154,7 +154,9 @@ private[spark] class TorrentBroadcast[T: ClassTag](
         checksums(i) = calcChecksum(block)
       }
       val pieceId = BroadcastBlockId(id, "piece" + i)
-      blockManager.persistBroadcastPiece(pieceId, block)
+      if (isExecutorSide) {
+        blockManager.persistBroadcastPiece(pieceId, block)
+      }
       val bytes = new ChunkedByteBuffer(block.duplicate())
       if (!blockManager.putBytes(pieceId, bytes, MEMORY_AND_DISK_SER, tellMaster = true)) {
         throw new SparkException(s"Failed to store $pieceId of $broadcastId in local BlockManager")
diff --git a/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcastFactory.scala b/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcastFactory.scala
@@ -42,7 +42,7 @@ private[spark] class TorrentBroadcastFactory extends BroadcastFactory {
   }
 
   override def uploadBroadcast[T: ClassTag](value_ : T, id: Long): Int = {
-    new TorrentBroadcast[T](value_, id).getNumBlocks
+    new TorrentBroadcast[T](value_, id, true).getNumBlocks
   }
 
   override def stop() { }
diff --git a/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala b/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala
@@ -205,6 +205,9 @@ private[spark] class DiskBlockManager(conf: SparkConf, deleteFilesOnStop: Boolea
       }
       try {
         val fs = dirPath.getFileSystem(SparkHadoopUtil.get.conf)
+        if (!fs.exists(dirPath)) {
+          return
+        }
         val files = fs.listStatus(dirPath, fileFilter).map(_.getPath)
         for (file <- files) {
           if (fs.exists(file)) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
@@ -1071,7 +1071,7 @@ class DatasetSuite extends QueryTest with SharedSQLContext {
 
       agg.queryExecution.executedPlan.collectFirst {
         case ShuffleExchange(_, _: RDDScanExec, _) =>
-        case BroadcastExchangeExec(_, _: RDDScanExec) =>
+        case BroadcastExchangeExec(_, _: RDDScanExec, _) =>
       }.foreach { _ =>
         fail(
           "No Exchange should be inserted above RDDScanExec since the checkpointed Dataset " +

Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,7 @@ private[spark] class TorrentBroadcastFactory extends BroadcastFactory {`
`42`	`42`	`}`
`43`	`43`
`44`	`44`	`override def uploadBroadcast[T: ClassTag](value_ : T, id: Long): Int = {`
`45`		`- new TorrentBroadcast[T](value_, id).getNumBlocks`
	`45`	`+ new TorrentBroadcast[T](value_, id, true).getNumBlocks`
`46`	`46`	`}`
`47`	`47`
`48`	`48`	`override def stop() { }`