cloud-fan commented on code in PR #59009:
URL: https://github.com/apache/spark/pull/59009#discussion_r4106950513
##########
core/src/test/java/org/apache/spark/unsafe/map/AbstractBytesToBytesMapSuite.java:
##########
@@ -875,4 +879,362 @@ public void freeAfterFailedReset() {
}
}
+ @Test
+ public void lookupAfterFreeDoesNotRestoreArray() {
+ memoryManager.limit(5000);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager, serializerManager,
256, 0.5, 4000);
+ final long[] key = new long[]{1L};
+ try {
+ memoryManager.markExecutionAsOutOfMemoryOnce();
+ assertThrows(SparkOutOfMemoryError.class, map::reset);
+ map.free();
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+
+ map.reset();
+ assertEquals(0, map.getNumDataPages());
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+
+ assertThrows(
+ IllegalStateException.class,
+ () -> map.lookup(key, Platform.LONG_ARRAY_OFFSET, 8));
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+ } finally {
+ map.free();
+ }
+ }
+
+ @Test
+ public void lookupAfterDestructiveIterationDoesNotRestoreArray() {
+ memoryManager.limit(5000);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager, serializerManager,
256, 0.5, 4000);
+ final long[] key = new long[]{1L};
+ try {
+ memoryManager.markExecutionAsOutOfMemoryOnce();
+ assertThrows(SparkOutOfMemoryError.class, map::reset);
+ map.destructiveIterator();
+ memoryManager.limit(PAGE_SIZE_BYTES);
+
+ assertThrows(
+ IllegalStateException.class,
+ () -> map.lookup(key, Platform.LONG_ARRAY_OFFSET, 8));
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+ } finally {
+ map.free();
+ }
+ }
+
+ @Test
+ public void resetAfterDestructiveIterationFails() {
+ memoryManager.limit(PAGE_SIZE_BYTES);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager, serializerManager,
256, 0.5, 4000);
+ try {
+ map.destructiveIterator();
+ assertThrows(IllegalStateException.class, map::reset);
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+ } finally {
+ map.free();
+ }
+ }
+
+ @Test
+ public void arrayAccessorsRecoverAfterFailedReset() {
+ memoryManager.limit(5000);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager, serializerManager,
256, 0.5, 4000);
+ try {
+ memoryManager.markExecutionAsOutOfMemoryOnce();
+ assertThrows(SparkOutOfMemoryError.class, map::reset);
+ memoryManager.limit(PAGE_SIZE_BYTES);
+ assertEquals(512L, map.getArray().size());
+
+ memoryManager.markExecutionAsOutOfMemoryOnce();
+ assertThrows(SparkOutOfMemoryError.class, map::reset);
+ assertEquals(256, map.maxNumKeysIndex());
+ } finally {
+ map.free();
+ }
+ }
+
+ @Test
+ public void reuseMapAfterFailedReset() {
+ // A reset() whose eager longArray reallocation OOMs leaves the map with
longArray == null (and,
Review Comment:
Addressed in 60eea7415ee. The stale pre-fix narrative was replaced with the
invariant being tested: failed reset clears old page state and lookup must not
allocate implicitly.
##########
core/src/test/java/org/apache/spark/unsafe/map/AbstractBytesToBytesMapSuite.java:
##########
@@ -875,4 +879,362 @@ public void freeAfterFailedReset() {
}
}
+ @Test
+ public void lookupAfterFreeDoesNotRestoreArray() {
+ memoryManager.limit(5000);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager, serializerManager,
256, 0.5, 4000);
+ final long[] key = new long[]{1L};
+ try {
+ memoryManager.markExecutionAsOutOfMemoryOnce();
+ assertThrows(SparkOutOfMemoryError.class, map::reset);
+ map.free();
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+
+ map.reset();
+ assertEquals(0, map.getNumDataPages());
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+
+ assertThrows(
+ IllegalStateException.class,
+ () -> map.lookup(key, Platform.LONG_ARRAY_OFFSET, 8));
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+ } finally {
+ map.free();
+ }
+ }
+
+ @Test
+ public void lookupAfterDestructiveIterationDoesNotRestoreArray() {
+ memoryManager.limit(5000);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager, serializerManager,
256, 0.5, 4000);
+ final long[] key = new long[]{1L};
+ try {
+ memoryManager.markExecutionAsOutOfMemoryOnce();
+ assertThrows(SparkOutOfMemoryError.class, map::reset);
+ map.destructiveIterator();
+ memoryManager.limit(PAGE_SIZE_BYTES);
+
+ assertThrows(
+ IllegalStateException.class,
+ () -> map.lookup(key, Platform.LONG_ARRAY_OFFSET, 8));
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+ } finally {
+ map.free();
+ }
+ }
+
+ @Test
+ public void resetAfterDestructiveIterationFails() {
+ memoryManager.limit(PAGE_SIZE_BYTES);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager, serializerManager,
256, 0.5, 4000);
+ try {
+ map.destructiveIterator();
+ assertThrows(IllegalStateException.class, map::reset);
+ assertEquals(0L, taskMemoryManager.getMemoryConsumptionForThisTask());
+ } finally {
+ map.free();
+ }
+ }
+
+ @Test
+ public void arrayAccessorsRecoverAfterFailedReset() {
+ memoryManager.limit(5000);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager, serializerManager,
256, 0.5, 4000);
+ try {
+ memoryManager.markExecutionAsOutOfMemoryOnce();
+ assertThrows(SparkOutOfMemoryError.class, map::reset);
+ memoryManager.limit(PAGE_SIZE_BYTES);
+ assertEquals(512L, map.getArray().size());
+
+ memoryManager.markExecutionAsOutOfMemoryOnce();
+ assertThrows(SparkOutOfMemoryError.class, map::reset);
+ assertEquals(256, map.maxNumKeysIndex());
+ } finally {
+ map.free();
+ }
+ }
+
+ @Test
+ public void reuseMapAfterFailedReset() {
+ // A reset() whose eager longArray reallocation OOMs leaves the map with
longArray == null (and,
+ // because the trailing reset() bookkeeping never runs, a stale
currentPage). A subsequent
+ // lookup then NPEs in safeLookup (this.longArray is null), and a later
append would write into
+ // a freed page. The map must instead recover to a usable empty state.
+ memoryManager.limit(5000);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager, serializerManager,
256, 0.5, 4000);
+ try {
+ final long[] key = new long[]{1L};
+ final long[] value = new long[]{42L};
+ BytesToBytesMap.Location loc = map.lookup(key,
Platform.LONG_ARRAY_OFFSET, 8);
+ loc.append(key, Platform.LONG_ARRAY_OFFSET, 8, value,
Platform.LONG_ARRAY_OFFSET, 8);
+
+ // Force OOM on the reallocation inside reset(), leaving longArray ==
null.
+ memoryManager.markExecutionAsOutOfMemoryOnce();
+ assertThrows(SparkOutOfMemoryError.class, map::reset);
+
+ // The map must recover: lookup must not NPE, it must report an empty
map, and it must be
+ // usable again (append into a fresh page, then read the value back).
+ memoryManager.limit(PAGE_SIZE_BYTES);
+ loc = map.lookup(key, Platform.LONG_ARRAY_OFFSET, 8);
+ Assertions.assertFalse(loc.isDefined());
+ Assertions.assertEquals(0, map.numValues());
+ Assertions.assertTrue(
+ loc.append(key, Platform.LONG_ARRAY_OFFSET, 8, value,
Platform.LONG_ARRAY_OFFSET, 8));
+ Assertions.assertEquals(1, map.getNumDataPages());
+ loc = map.lookup(key, Platform.LONG_ARRAY_OFFSET, 8);
+ Assertions.assertTrue(loc.isDefined());
+ Assertions.assertEquals(42L, Platform.getLong(loc.getValueBase(),
loc.getValueOffset()));
+ Assertions.assertEquals(1, map.numValues());
+ } finally {
+ map.free();
+ }
+ }
+
+ @Test
+ public void reuseMapWithSemanticKeyOperationsAfterFailedReset() {
+ memoryManager.limit(5000);
+ BytesToBytesMap.KeyOperationsFactory caseInsensitiveWithConstantHash =
Review Comment:
Addressed in 60eea7415ee. The duplicated semantic-key case and repeated
failed-reset fixtures were removed. Coverage is consolidated into two
failed-reset tests plus focused terminal-state tests, reducing the added test
code to 141 lines.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -140,6 +140,7 @@ public interface KeyOperationsFactory {
* while position {@code 2 * i + 1} in the array holds key's full 32-bit
hashcode.
*/
@Nullable private LongArray longArray;
+
Review Comment:
Addressed in 60eea7415ee. The extra blank line is removed, so the TODO is
directly below longArray again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
