dongjoon-hyun commented on code in PR #60167:
URL: https://github.com/apache/spark/pull/60167#discussion_r4498154493
##########
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:
**1. [Low] A deserialized map still reserves at least 1 MiB, while 4.1
allocated what it read**
The description says that each map now reserves memory close to what it
stores, as in Spark 4.1 and earlier. That holds for a map built in a task, but
not for a deserialized one. `read()`
([L979](https://github.com/apache/spark/blob/4331d42b9e96d476a97d04464111d369c9d92dbb/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala#L979))
calls `allocatePage(pageLength * 8L)`, which the new floor rounds up to 1 MiB.
In 4.1, `read()` allocated exactly `pageLength` words ([v4.1.0
L976](https://github.com/apache/spark/blob/v4.1.0/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala#L976),
tagged 2025-12-11).
For a broadcast `LongHashedRelation` with 10 rows, the page holds 240 bytes.
Each executor that reads the broadcast now keeps a 1 MiB `long[]` for it.
`estimatedSize` reports 1 MiB too, and the memory store uses it for the
broadcast block, and BHJ adds it to the peak execution memory metric. That is
much better than the 64 MiB before this PR. Still, a query that broadcasts many
small dimension tables pays 1 MiB for each of them on every executor.
One way to match 4.1 would be to pass a floor of 8 bytes to
`MemoryConsumer`, and keep the explicit 1 MiB request in `init()`. `grow()`
already asks for at least twice the current page, so only `read()` would behave
differently: it would allocate what it reads. (A floor of 0 would not work,
since `allocatePageInternal` returns `null` when it acquires 0 bytes, which an
empty map would hit.) If the 1 MiB floor is intended, could the description say
that deserialized maps differ from 4.1 here?
##########
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:
**3. [Nit] The comment in `grow()` still says `allocatePage(0)` falls back
to the default page size**
With this line, `allocatePage` rounds a request up to `MIN_PAGE_SIZE_BYTES`,
not to the default page size. The comment at
[L829-L834](https://github.com/apache/spark/blob/4331d42b9e96d476a97d04464111d369c9d92dbb/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala#L829-L834)
still describes the old fallback ("causing `allocatePage(0)` to fall back to
the default page size"). Could it name the 1 MiB floor instead?
##########
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:
**2. [Low] Power-of-two on-heap pages no longer fit G1 regions**
On heap, a page of `n` bytes is a `long[]` of `n + 16` bytes with the array
header. With exact powers of two (1 MiB, 2 MiB, 4 MiB, ...), a page at least as
large as a G1 region becomes a humongous object that spans one extra region.
With an 8 GiB heap (4 MiB regions), a 4 MiB page takes 8 MiB of heap, and a 64
MiB page takes 68 MiB.
SPARK-37593 avoids this for the default page size:
`MemoryManager.defaultPageSizeBytes` subtracts `Platform.LONG_ARRAY_OFFSET`
under G1, ZGC and Shenandoah in on-heap mode ([MemoryManager.scala
L275-L280](https://github.com/apache/spark/blob/4331d42b9e96d476a97d04464111d369c9d92dbb/core/src/main/scala/org/apache/spark/memory/MemoryManager.scala#L275-L280)).
Before this PR, the first page had that size (64 MiB minus 16 bytes), and
doubling kept every page aligned.
4.1 allocated the same power-of-two arrays, so this is not a regression from
4.1. Since this PR picks the page sizes anyway, would it be worth starting at
`(1L << 20) - Platform.LONG_ARRAY_OFFSET` in on-heap mode? Doubling
`page.size() / 8` words then keeps every array, header included, within a power
of two. This needs a floor below the initial page size, as in item 1.
##########
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:
**4. [Nit] The deserialized map does not use the test's memory manager**
The deserialized map comes from the no-arg constructor
([`HashedRelation.scala`
L577-L587](https://github.com/apache/spark/blob/4331d42b9e96d476a97d04464111d369c9d92dbb/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala#L577-L587)),
which builds its own `TaskMemoryManager` on `Runtime.getRuntime.maxMemory`. So
the 64 MiB page size of `largeUmm` applies only to `originalMap`. For
`deserializedMap`, the test depends on the page size derived from the JVM heap,
which the comment at L322-L323 says these tests avoid.
The test still catches the regression, since that derived size is 64 MiB
once the heap is over 1 GiB. But L417 checks only `largeUmm`, so it does not
show that `deserializedMap.free()` released its memory. Could the comment be
narrowed to the maps built on `largeMm`, and the deserialized map be checked
with `getUsed === 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
Review Comment:
**6. [Nit] The `built` flag could be replaced by a single `finally`**
The maps are freed in two places: in the `finally` when the build fails, and
at L388 after the checks. If a check at L383-L387 fails, the maps are not
freed. Since `smallUmm` is local, that does not affect other tests. Still,
moving the checks into the `try` and freeing in the `finally` unconditionally
would drop the flag and the second `free()`:
```scala
try {
(0 until 16).foreach { i => ... }
maps.zipWithIndex.foreach { case (map, i) => ... }
} 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 {
maps.foreach(_.free())
}
assert(smallUmm.executionMemoryUsed === 0)
```
##########
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:
**7. [Nit] The companion object is narrower than the class**
The class is `private[execution]`, and `AggregateBenchmark` and
`HashedRelationMetricsBenchmark` in `execution.benchmark` already create it.
They cannot read `MIN_PAGE_SIZE_BYTES` from this `private[joins]` object,
though. Making the object `private[execution]` would match the class.
##########
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:
**5. [Nit] Only the Java serialization path is tested**
`sparkContext.env.serializer` is the `JavaSerializer` in this suite, so this
test goes through `readExternal` only. A broadcast with Kryo goes through
`read(kryo, in)`. Both call the same private `read` today, so this is only
about keeping the Kryo path covered if they ever diverge. Running the same loop
with a `KryoSerializer` as well, as `Spark-14521` does, would cover both.
--
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]
