diff options
Diffstat (limited to 'sql')
2 files changed, 4 insertions, 13 deletions
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java b/sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java index b81f67a16b..9df5780e4f 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java @@ -85,7 +85,7 @@ public final class UnsafeKVExternalSorter { // We will use the number of elements in the map as the initialSize of the // UnsafeInMemorySorter. Because UnsafeInMemorySorter does not accept 0 as the initialSize, // we will use 1 as its initial size if the map is empty. - // TODO: track pointer array memory used by this in-memory sorter! + // TODO: track pointer array memory used by this in-memory sorter! (SPARK-10474) final UnsafeInMemorySorter inMemSorter = new UnsafeInMemorySorter( taskMemoryManager, recordComparator, prefixComparator, Math.max(1, map.numElements())); @@ -124,13 +124,8 @@ public final class UnsafeKVExternalSorter { pageSizeBytes, inMemSorter); - // Note: This spill doesn't actually release any memory, so if we try to allocate a new - // pointer array immediately after the spill then we may fail to acquire sufficient space - // for it (SPARK-10474). For this reason, we must initialize for writing explicitly *after* - // we have actually freed memory from our map. - sorter.spill(false /* initialize for writing */); + sorter.spill(); map.free(); - sorter.initializeForWriting(); } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/UnsafeFixedWidthAggregationMapSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/UnsafeFixedWidthAggregationMapSuite.scala index ada4d42f99..1739798a24 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/UnsafeFixedWidthAggregationMapSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/UnsafeFixedWidthAggregationMapSuite.scala @@ -200,9 +200,7 @@ class UnsafeFixedWidthAggregationMapSuite val sorter = map.destructAndCreateExternalSorter() withClue(s"destructAndCreateExternalSorter should release memory used by the map") { - // 4096 * 16 is the initial size allocated for the pointer/prefix array in the in-mem sorter. - assert(shuffleMemoryManager.getMemoryConsumptionForThisTask() === - initialMemoryConsumption + 4096 * 16) + assert(shuffleMemoryManager.getMemoryConsumptionForThisTask() === initialMemoryConsumption) } // Add more keys to the sorter and make sure the results come out sorted. @@ -305,9 +303,7 @@ class UnsafeFixedWidthAggregationMapSuite val sorter = map.destructAndCreateExternalSorter() withClue(s"destructAndCreateExternalSorter should release memory used by the map") { - // 4096 * 16 is the initial size allocated for the pointer/prefix array in the in-mem sorter. - assert(shuffleMemoryManager.getMemoryConsumptionForThisTask() === - initialMemoryConsumption + 4096 * 16) + assert(shuffleMemoryManager.getMemoryConsumptionForThisTask() === initialMemoryConsumption) } // Add more keys to the sorter and make sure the results come out sorted. |