viirya commented on code in PR #60167:
URL: https://github.com/apache/spark/pull/60167#discussion_r4499004831
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala:
##########
@@ -990,6 +991,20 @@ private[execution] final class LongToUnsafeRowMap(
}
}
+private[joins] object LongToUnsafeRowMap {
+ /**
+ * The page size floor passed to `MemoryConsumer`, which is also the size of
the initial page.
+ *
+ * `MemoryConsumer.allocatePage` rounds every request up to the consumer's
page size. With the
+ * default page size (`spark.buffer.pageSize`, up to 64 MiB), even an empty
map would reserve
+ * 64 MiB of execution memory. A task that keeps many maps alive at once,
e.g. a stage with a
+ * chain of shuffled hash joins, then runs out of execution memory long
before the maps hold
+ * that much data. With a 1 MiB floor, the page starts at 1 MiB and grows by
doubling, so the
+ * memory each map reserves stays close to what it stores.
Review Comment:
Good catch, thanks. Fixed in c6b2e5b0b7a as you suggested: `MemoryConsumer`
now gets a floor of one word (`MIN_PAGE_SIZE_BYTES = 8`), and `init()` requests
the 1 MiB initial page explicitly (`INITIAL_PAGE_SIZE_BYTES`). `grow()` always
asks for at least twice the current page, so only `read()` changes: it
allocates exactly what it reads, as in 4.1. An empty map's page is 8 bytes. I
also updated the PR description.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala:
##########
@@ -990,6 +991,20 @@ private[execution] final class LongToUnsafeRowMap(
}
}
+private[joins] object LongToUnsafeRowMap {
+ /**
+ * The page size floor passed to `MemoryConsumer`, which is also the size of
the initial page.
+ *
+ * `MemoryConsumer.allocatePage` rounds every request up to the consumer's
page size. With the
+ * default page size (`spark.buffer.pageSize`, up to 64 MiB), even an empty
map would reserve
+ * 64 MiB of execution memory. A task that keeps many maps alive at once,
e.g. a stage with a
+ * chain of shuffled hash joins, then runs out of execution memory long
before the maps hold
+ * that much data. With a 1 MiB floor, the page starts at 1 MiB and grows by
doubling, so the
+ * memory each map reserves stays close to what it stores.
+ */
+ val MIN_PAGE_SIZE_BYTES: Long = 1L << 20
Review Comment:
Agreed, the analysis is right. As you noted, 4.1 allocated the same
power-of-two arrays, so this is not a regression, and it only matters for pages
at least as large as a G1 region. I'd like to keep this PR to restoring the 4.1
behavior, since it fixes a 4.2.0 regression and should be easy to backport.
Starting at `(1L << 20) - Platform.LONG_ARRAY_OFFSET` would make the page
sizes, and the exact sizes in the tests, depend on the GC and the memory mode.
I can do it in a follow-up if you think it's worth it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala:
##########
@@ -537,7 +537,8 @@ private[execution] final class LongToUnsafeRowMap(
val mm: TaskMemoryManager,
capacity: Int,
ignoresDuplicatedKey: Boolean = false)
- extends MemoryConsumer(mm, mm.getTungstenMemoryMode) with Externalizable
with KryoSerializable {
+ extends MemoryConsumer(mm, LongToUnsafeRowMap.MIN_PAGE_SIZE_BYTES,
mm.getTungstenMemoryMode)
Review Comment:
Updated in c6b2e5b0b7a. The comment now says that `allocatePage(0)` would
return a page of only `MIN_PAGE_SIZE_BYTES`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala:
##########
@@ -318,6 +318,105 @@ abstract class HashedRelationSuite extends
SharedSparkSession {
secondTimeSerialized.free()
}
+ /**
+ * Creates a memory manager whose page size is set explicitly, so that the
tests below do not
+ * depend on the page size derived from the JVM heap and the number of cores.
+ */
+ private def newMemoryManager(pageSize: Long, memorySize: Long):
UnifiedMemoryManager = {
+ new UnifiedMemoryManager(
+ new SparkConf()
+ .set(MEMORY_OFFHEAP_ENABLED, useOffHeapMemoryMode)
+ .set(MEMORY_OFFHEAP_SIZE, memorySize)
+ .set(BUFFER_PAGESIZE, pageSize),
+ memorySize,
+ memorySize / 2,
+ 1)
+ }
+
+ test("SPARK-60998: LongToUnsafeRowMap does not round its page up to the page
size") {
+ val largeUmm = newMemoryManager(ByteUnit.MiB.toBytes(64),
ByteUnit.GiB.toBytes(1))
+ val largeMm = new TaskMemoryManager(largeUmm, 0)
+ assert(largeMm.pageSizeBytes() === ByteUnit.MiB.toBytes(64))
+
+ // Capacity 16 gives an array of 32 longs, which does not grow for a
single key.
+ val arrayBytes = 32 * 8L
+ val map = new LongToUnsafeRowMap(largeMm, 16)
+ assert(map.getMode === largeMm.getTungstenMemoryMode)
+ assert(map.getUsed === arrayBytes + LongToUnsafeRowMap.MIN_PAGE_SIZE_BYTES)
+
+ // A value larger than the initial page doubles the page to 2 MiB, not 64
MiB.
+ val unsafeProj = UnsafeProjection.create(Array[DataType](StringType))
+ val bigStr = UTF8String.fromString("x".repeat(3 << 19))
+ map.append(0L, unsafeProj(InternalRow(bigStr)))
+ assert(map.getUsed === arrayBytes + 2 *
LongToUnsafeRowMap.MIN_PAGE_SIZE_BYTES)
+
+ val resultRow = new UnsafeRow(1)
+ assert(map.getValue(0L, resultRow).getUTF8String(0) === bigStr)
+ map.free()
+ assert(largeUmm.executionMemoryUsed === 0)
+ }
+
+ test("SPARK-60998: many LongToUnsafeRowMaps fit in one task when each holds
little data") {
+ // 16 maps rounded up to a 64 MiB page each would need 1 GiB, 4x the
memory available.
+ val smallUmm = newMemoryManager(ByteUnit.MiB.toBytes(64),
ByteUnit.MiB.toBytes(256))
+ val smallMm = new TaskMemoryManager(smallUmm, 0)
+ val unsafeProj = UnsafeProjection.create(Seq(BoundReference(0, LongType,
false)))
+ val maps = ArrayBuffer.empty[LongToUnsafeRowMap]
+ var built = false
+ try {
+ (0 until 16).foreach { i =>
+ val map = new LongToUnsafeRowMap(smallMm, 16)
+ maps += map
+ (0 until 10).foreach(k => map.append(k,
unsafeProj(InternalRow(k.toLong + i))))
+ map.optimize()
+ }
+ built = true
+ } catch {
+ // SparkOutOfMemoryError is fatal to ScalaTest and would abort the whole
suite.
+ case e: SparkOutOfMemoryError =>
+ fail("Could not build 16 small LongToUnsafeRowMaps in one task", e)
+ } finally {
+ if (!built) {
+ maps.foreach(_.free())
+ }
+ }
+ maps.zipWithIndex.foreach { case (map, i) =>
+ assert(map.getUsed < 2 * LongToUnsafeRowMap.MIN_PAGE_SIZE_BYTES)
+ val resultRow = new UnsafeRow(1)
+ (0 until 10).foreach(k => assert(map.getValue(k, resultRow).getLong(0)
=== k + i))
+ }
+ maps.foreach(_.free())
+ assert(smallUmm.executionMemoryUsed === 0)
+ }
+
+ test("SPARK-60998: deserialized LongToUnsafeRowMap does not round its page
up to the page size") {
+ val largeUmm = newMemoryManager(ByteUnit.MiB.toBytes(64),
ByteUnit.GiB.toBytes(1))
+ val largeMm = new TaskMemoryManager(largeUmm, 0)
+ val unsafeProj = UnsafeProjection.create(Array[DataType](LongType))
+ val ser = sparkContext.env.serializer.newInstance()
+
+ // Both a non-empty map and an empty map, whose serialized page has length
0.
+ Seq(10, 0).foreach { numRows =>
+ val originalMap = new LongToUnsafeRowMap(largeMm, 16)
+ (0 until numRows).foreach(k => originalMap.append(k,
unsafeProj(InternalRow(k * 2L))))
+ if (numRows > 0) {
+ originalMap.optimize()
+ }
+ val deserializedMap =
ser.deserialize[LongToUnsafeRowMap](ser.serialize(originalMap))
Review Comment:
Right, the deserialized map gets its own memory manager from the no-arg
constructor. Fixed in c6b2e5b0b7a: the `newMemoryManager` comment now refers
only to the maps built on it, and the test checks the exact memory of the
deserialized map (its array plus the page it read) and that `getUsed` is 0
after `free()`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala:
##########
@@ -318,6 +318,105 @@ abstract class HashedRelationSuite extends
SharedSparkSession {
secondTimeSerialized.free()
}
+ /**
+ * Creates a memory manager whose page size is set explicitly, so that the
tests below do not
+ * depend on the page size derived from the JVM heap and the number of cores.
+ */
+ private def newMemoryManager(pageSize: Long, memorySize: Long):
UnifiedMemoryManager = {
+ new UnifiedMemoryManager(
+ new SparkConf()
+ .set(MEMORY_OFFHEAP_ENABLED, useOffHeapMemoryMode)
+ .set(MEMORY_OFFHEAP_SIZE, memorySize)
+ .set(BUFFER_PAGESIZE, pageSize),
+ memorySize,
+ memorySize / 2,
+ 1)
+ }
+
+ test("SPARK-60998: LongToUnsafeRowMap does not round its page up to the page
size") {
+ val largeUmm = newMemoryManager(ByteUnit.MiB.toBytes(64),
ByteUnit.GiB.toBytes(1))
+ val largeMm = new TaskMemoryManager(largeUmm, 0)
+ assert(largeMm.pageSizeBytes() === ByteUnit.MiB.toBytes(64))
+
+ // Capacity 16 gives an array of 32 longs, which does not grow for a
single key.
+ val arrayBytes = 32 * 8L
+ val map = new LongToUnsafeRowMap(largeMm, 16)
+ assert(map.getMode === largeMm.getTungstenMemoryMode)
+ assert(map.getUsed === arrayBytes + LongToUnsafeRowMap.MIN_PAGE_SIZE_BYTES)
+
+ // A value larger than the initial page doubles the page to 2 MiB, not 64
MiB.
+ val unsafeProj = UnsafeProjection.create(Array[DataType](StringType))
+ val bigStr = UTF8String.fromString("x".repeat(3 << 19))
+ map.append(0L, unsafeProj(InternalRow(bigStr)))
+ assert(map.getUsed === arrayBytes + 2 *
LongToUnsafeRowMap.MIN_PAGE_SIZE_BYTES)
+
+ val resultRow = new UnsafeRow(1)
+ assert(map.getValue(0L, resultRow).getUTF8String(0) === bigStr)
+ map.free()
+ assert(largeUmm.executionMemoryUsed === 0)
+ }
+
+ test("SPARK-60998: many LongToUnsafeRowMaps fit in one task when each holds
little data") {
+ // 16 maps rounded up to a 64 MiB page each would need 1 GiB, 4x the
memory available.
+ val smallUmm = newMemoryManager(ByteUnit.MiB.toBytes(64),
ByteUnit.MiB.toBytes(256))
+ val smallMm = new TaskMemoryManager(smallUmm, 0)
+ val unsafeProj = UnsafeProjection.create(Seq(BoundReference(0, LongType,
false)))
+ val maps = ArrayBuffer.empty[LongToUnsafeRowMap]
+ var built = false
+ try {
+ (0 until 16).foreach { i =>
+ val map = new LongToUnsafeRowMap(smallMm, 16)
+ maps += map
+ (0 until 10).foreach(k => map.append(k,
unsafeProj(InternalRow(k.toLong + i))))
+ map.optimize()
+ }
+ built = true
+ } catch {
+ // SparkOutOfMemoryError is fatal to ScalaTest and would abort the whole
suite.
+ case e: SparkOutOfMemoryError =>
+ fail("Could not build 16 small LongToUnsafeRowMaps in one task", e)
+ } finally {
+ if (!built) {
+ maps.foreach(_.free())
+ }
+ }
+ maps.zipWithIndex.foreach { case (map, i) =>
+ assert(map.getUsed < 2 * LongToUnsafeRowMap.MIN_PAGE_SIZE_BYTES)
+ val resultRow = new UnsafeRow(1)
+ (0 until 10).foreach(k => assert(map.getValue(k, resultRow).getLong(0)
=== k + i))
+ }
+ maps.foreach(_.free())
+ assert(smallUmm.executionMemoryUsed === 0)
+ }
+
+ test("SPARK-60998: deserialized LongToUnsafeRowMap does not round its page
up to the page size") {
+ val largeUmm = newMemoryManager(ByteUnit.MiB.toBytes(64),
ByteUnit.GiB.toBytes(1))
+ val largeMm = new TaskMemoryManager(largeUmm, 0)
+ val unsafeProj = UnsafeProjection.create(Array[DataType](LongType))
+ val ser = sparkContext.env.serializer.newInstance()
Review Comment:
Added in c6b2e5b0b7a. The test now runs the same loop with a
`JavaSerializer` and a `KryoSerializer` (reference tracking off, as in
`Spark-14521`).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala:
##########
@@ -318,6 +318,105 @@ abstract class HashedRelationSuite extends
SharedSparkSession {
secondTimeSerialized.free()
}
+ /**
+ * Creates a memory manager whose page size is set explicitly, so that the
tests below do not
+ * depend on the page size derived from the JVM heap and the number of cores.
+ */
+ private def newMemoryManager(pageSize: Long, memorySize: Long):
UnifiedMemoryManager = {
+ new UnifiedMemoryManager(
+ new SparkConf()
+ .set(MEMORY_OFFHEAP_ENABLED, useOffHeapMemoryMode)
+ .set(MEMORY_OFFHEAP_SIZE, memorySize)
+ .set(BUFFER_PAGESIZE, pageSize),
+ memorySize,
+ memorySize / 2,
+ 1)
+ }
+
+ test("SPARK-60998: LongToUnsafeRowMap does not round its page up to the page
size") {
+ val largeUmm = newMemoryManager(ByteUnit.MiB.toBytes(64),
ByteUnit.GiB.toBytes(1))
+ val largeMm = new TaskMemoryManager(largeUmm, 0)
+ assert(largeMm.pageSizeBytes() === ByteUnit.MiB.toBytes(64))
+
+ // Capacity 16 gives an array of 32 longs, which does not grow for a
single key.
+ val arrayBytes = 32 * 8L
+ val map = new LongToUnsafeRowMap(largeMm, 16)
+ assert(map.getMode === largeMm.getTungstenMemoryMode)
+ assert(map.getUsed === arrayBytes + LongToUnsafeRowMap.MIN_PAGE_SIZE_BYTES)
+
+ // A value larger than the initial page doubles the page to 2 MiB, not 64
MiB.
+ val unsafeProj = UnsafeProjection.create(Array[DataType](StringType))
+ val bigStr = UTF8String.fromString("x".repeat(3 << 19))
+ map.append(0L, unsafeProj(InternalRow(bigStr)))
+ assert(map.getUsed === arrayBytes + 2 *
LongToUnsafeRowMap.MIN_PAGE_SIZE_BYTES)
+
+ val resultRow = new UnsafeRow(1)
+ assert(map.getValue(0L, resultRow).getUTF8String(0) === bigStr)
+ map.free()
+ assert(largeUmm.executionMemoryUsed === 0)
+ }
+
+ test("SPARK-60998: many LongToUnsafeRowMaps fit in one task when each holds
little data") {
+ // 16 maps rounded up to a 64 MiB page each would need 1 GiB, 4x the
memory available.
+ val smallUmm = newMemoryManager(ByteUnit.MiB.toBytes(64),
ByteUnit.MiB.toBytes(256))
+ val smallMm = new TaskMemoryManager(smallUmm, 0)
+ val unsafeProj = UnsafeProjection.create(Seq(BoundReference(0, LongType,
false)))
+ val maps = ArrayBuffer.empty[LongToUnsafeRowMap]
+ var built = false
Review Comment:
Done in c6b2e5b0b7a. The checks moved into the `try`, and the `finally`
frees the maps unconditionally.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala:
##########
@@ -990,6 +991,20 @@ private[execution] final class LongToUnsafeRowMap(
}
}
+private[joins] object LongToUnsafeRowMap {
Review Comment:
Changed to `private[execution]` in c6b2e5b0b7a.
--
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]
