Fixed SPARK-1194 https://spark-project.atlassian.net/browse/SPARK-1194

holdenk · Mar 7, 2014 · 62c92ac · 62c92ac
1 parent 328c73d
commit 62c92ac
Showing 1 changed file with 16 additions and 6 deletions.
diff --git a/core/src/main/scala/org/apache/spark/storage/MemoryStore.scala b/core/src/main/scala/org/apache/spark/storage/MemoryStore.scala
@@ -236,13 +236,23 @@ private class MemoryStore(blockManager: BlockManager, maxMemory: Long)
         while (maxMemory - (currentMemory - selectedMemory) < space && iterator.hasNext) {
           val pair = iterator.next()
           val blockId = pair.getKey
-          if (rddToAdd.isDefined && rddToAdd == getRddId(blockId)) {
-            logInfo("Will not store " + blockIdToAdd + " as it would require dropping another " +
-              "block from the same RDD")
-            return false
+          // Apply the same-RDD rule for cache replacement. Quoted from the
+          // original RDD paper:
+          //
+          //    When a new RDD partition is computed but there is not enough
+          //    space to store it, we evict a partition from the least recently
+          //    accessed RDD, unless this is the same RDD as the one with the
+          //    new partition. In that case, we keep the old partition in memory
+          //    to prevent cycling partitions from the same RDD in and out.
+          //
+          // TODO implement LRU eviction
+          rddToAdd match {
+            case Some(rddId) if rddId == getRddId(blockId) =>
+              // no-op
+            case _ =>
+              selectedBlocks += blockId
+              selectedMemory += pair.getValue.size
           }
-          selectedBlocks += blockId
-          selectedMemory += pair.getValue.size
         }
       }