spark-test commented on code in PR #60196:
URL: https://github.com/apache/spark/pull/60196#discussion_r4500951214
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -1037,18 +1095,20 @@ public boolean append(Object kbase, long koff, int
klen, Object vbase, long voff
* @return whether there is enough space to allocate the new page.
*/
private boolean acquireNewPage(long required) {
+ final long size = Math.max(nextPageSizeBytes, required);
final MemoryBlock page;
try {
- page = allocatePage(required);
+ page = allocatePage(size, required);
} catch (SparkOutOfMemoryError e) {
return false;
}
// Retaining exact-fit minimum-retry pages would consume one page-table
slot per map entry.
- if (required < pageSizeBytes && page.size() == required &&
+ if (required < size && page.size() == required &&
isPageAllocationFromMinimumRetry(page)) {
freePage(page);
return false;
}
+ nextPageSizeBytes = Math.min(pageSizeBytes, nextPageSizeBytes * 2);
Review Comment:
**Non-blocking (P2):** With `INITIAL_PAGE_SIZE_BYTES` at 1 MiB, this
doubling gives an `UnsafeHashedRelation` growth pages of 1, 2, 4, 8, 16 and 32
MiB, which add up to 63 MiB, 1 MiB short of a 64 MiB page. Every later page is
a full page, so each full-page boundary now sits about 1 MiB below where it was
on the base branch, where all pages were the page size P.
A relation whose data falls in that last MiB below a multiple of P now needs
one more full page than before: about 63.5 MiB of rows reserves 127 MiB instead
of 64 MiB, and 127.5 MiB reserves 191 MiB instead of 128 MiB (from a model of
this code with 80-byte rows). That is 1 MiB / P of all relation sizes: about
1.6% at 64 MiB, but 25% with a 4 MiB page and half with a 2 MiB page, both
derived defaults on small executors. A configured page size that is not a
power-of-two multiple of 1 MiB fares worse, because the growth pages overshoot
one page: at 48 MiB about half of all sizes reserve up to 15 MiB more. Rows of
a few MiB hit it for most sizes.
These relations already reserved memory close to their size on the base
branch, so for them the change makes things worse rather than better. In a
shuffled hash join the extra full-page request can spill co-located consumers
or take the task's remaining memory, and a memory-limited task holding several
such relations can fail where it used to fit. For a broadcast it adds driver
and executor heap and inflates the `dataSize` that the broadcast size limit
checks. The excess is bounded per relation.
**Recommended change:** Cap each growth page of a growing BytesToBytesMap at
what is left to reach the page size, so the growth pages add up to exactly one
page size before full pages begin. Keep a running total of the page sizes the
map requested for the pages it kept; while that total is below the page size,
request the smaller of nextPageSizeBytes and the remainder, and from then on
the page size, still using a record's own size when it is larger. Reset the
total in reset(). At the default this keeps the 1, 2, 4, 8, 16 and 32 MiB pages
and adds a final 1 MiB page before the 64 MiB pages; with a 48 MiB page it
gives 1, 2, 4, 8, 16 and 17 MiB. Update the docs that describe the growth (the
nextPageSizeBytes field, the growing constructor, and
UnsafeHashedRelation.INITIAL_PAGE_SIZE_BYTES) and the core page-sequence test,
and add core map tests for page-size boundaries.
**Why this works:** The doubling pages stop at exactly one page size,
because the last growth page is only as large as what is left to reach it. From
then on the total reserved is a whole multiple of the page size, the same grid
as a layout of full pages from the start, instead of falling about 1 MiB short
of it or overshooting it. Counting requested sizes rather than granted ones
keeps maps created through the existing constructors, whose first request is
already the page size, on full-page requests even after a partial grant.
**Scope:** Cap the doubling pages of a growing BytesToBytesMap at what is
left to reach the page size, update the docs that describe the growth, and
update and add core tests for the page sequence and the page-size boundaries.
**Compatibility:** Maps created through the existing constructors, including
hash aggregation, keep requesting the page size for every page. Growing maps
keep doubling from the 1 MiB initial page, so at power-of-two page sizes a
relation smaller than the page size minus that page reserves what it does now,
and deserialized relations rebuild their pages with the same rule.
**Risks:** A relation larger than one page size reserves 1 MiB more than
with the current doubling, the same as the base branch, because its growth
pages now add up to a full page, and at the default a large relation uses seven
growth pages instead of six. Restoring the uncapped doubling rolls the change
back. When a record larger than the next page lands late in the growth phase,
the last growth page can be very small, down to a single record, which costs
one more page-table entry.
**Constraints:** Maps created through the existing constructors must keep
requesting the page size for every page, also after a partial grant, so hash
aggregation and the other existing users stay unchanged. A record larger than
the next page must still get a page of its own size, and the exact-fit
minimum-retry rejection must keep applying whenever the request was padded
beyond the record's need. reset() must restore both the next page size and the
running total, so a reset map starts again from the initial page size. If a
header adjustment is adopted for the heap-region question, apply it inside this
rule, so the growth pages still end within a few header bytes of one page size
instead of about 1 MiB short of it.
**Exclusions:**
- Rows of hundreds of KiB or more can still need one extra full page when
the data falls just below a multiple of the page size, within a window of up to
one row per growth page, because each growth page can leave up to one row of
unused tail. Closing that would need moving records or sizing pages to the row
size.
- A partial grant during the growth phase leaves the growth pages short of
one page size, because the total counts the requested size. This happens only
under memory pressure, and counting granted bytes instead would change the
requests of the existing constructors.
**Success:** For a map created with an initial page size, when no record
needs a page of its own and every request is granted in full, the data pages
before the first full page add up to exactly the page size: the doubling pages,
with the last one cut to what is left to reach the page size. An
UnsafeHashedRelation of small rows never reserves more data-page memory than a
layout of full pages from the start, except within a window of at most one row
per growth page just below each multiple of the page size; for example, about
63.5 MiB of 80-byte rows reserves 64 MiB, not 127 MiB. With a page size that is
a power-of-two multiple of the initial size, a relation whose data stays below
the page size minus the initial size reserves exactly what it does with the
current doubling. Maps created through the existing constructors still request
the page size for every page, also after a partial grant, so hash aggregation
behaves as before. reset() restarts the page sequence from the initial p
age size.
**Verification:**
- **Regression:** A map whose initial page size is a small fraction of its
page size, filled with small records until its data lies within the last
initial page below one page size but more than one record per growth page below
it, and then likewise below two page sizes, reserves exactly one and then two
page sizes of data pages, not a page size plus the growth pages.
- **Regression:** When the page size is not a power-of-two multiple of the
initial size, for example a 6 KiB page with a 1 KiB initial page, the growth
pages are 1, 2 and 3 KiB and end exactly at the page size before full pages
begin, and a map with about 5 KiB of data reserves 6 KiB, not 7 KiB.
- **Behavior:** The page-sequence test checks the first page, the following
doubling pages, the last growth page cut to what is left to reach the page
size, an exact-size page for a record larger than the next page, and the
initial page size again after reset().
- **Compatibility:** Maps created through the existing constructors still
request the page size for every page, including after a partial first page, and
the existing BytesToBytesMap, TaskMemoryManager and hash aggregation tests pass
unchanged.
- **Behavior:** Small UnsafeHashedRelations still reserve less than two
initial pages, both when built and after Java and Kryo deserialization.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -259,10 +290,37 @@ public BytesToBytesMap(
throw new IllegalArgumentException("Page size " + pageSizeBytes + "
cannot exceed " +
TaskMemoryManager.MAXIMUM_PAGE_SIZE_BYTES);
}
+ if (initialPageSizeBytes <= 0) {
+ throw new IllegalArgumentException("Initial page size must be greater
than 0");
+ }
+ this.initialPageSizeBytes = Math.min(initialPageSizeBytes, pageSizeBytes);
Review Comment:
**Nit (P3):** `UnsafeHashedRelation` relies on this cap whenever the
`SparkEnv` page size is below 1 MiB, from an explicit `spark.buffer.pageSize`
or the 1 MiB - 16 bytes derived default under G1 on-heap. No test exercises it:
`dataPagesGrowFromInitialPageSize` uses a 1 KiB initial page with a 4 KiB page
size, and the existing constructors pass equal values. If the cap regressed,
such a relation would reserve up to 1 MiB, more than the single page it used to
reserve, and no test would fail. The existing `appendUntilNewDataPage` helper
can check the first page of such a map.
**Verification:**
- **Regression:** A map created with an initial page size larger than its
page size allocates its first data page at the page size, and the check fails
if that first page can exceed the page size.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -175,6 +175,17 @@ private enum MapState {
*/
private final long pageSizeBytes;
+ /**
+ * The size of the first data page, at most {@link #pageSizeBytes}.
Review Comment:
**Nit (P3):** This doc and the `nextPageSizeBytes` doc below it link to
`pageSizeBytes` as the cap of a growing map's pages, but the unchanged
`pageSizeBytes` doc just above still says it is "The size of the data pages
that hold key and value data". That is no longer true for any
`UnsafeHashedRelation` map, so the field docs now contradict each other and
invite reading `pageSizeBytes` as the size of every page. That doc could
instead call it the maximum size of regular data pages, which a map with a
smaller initial page grows up to; its claim that this limits the maximum entry
size was already inaccurate, since a larger record gets a page of its own size.
**Verification:**
- **Inspection:** The field doc names pageSizeBytes as the maximum size of
regular data pages, and no longer says it is the size of every data page or the
maximum entry size.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala:
##########
@@ -452,6 +453,17 @@ private[joins] class UnsafeHashedRelation(
private[joins] object UnsafeHashedRelation {
+ /**
+ * The size of the first data page of the map. Later pages double in size up
to the page size.
+ *
+ * Otherwise `BytesToBytesMap` rounds its first page up to the page size
(`spark.buffer.pageSize`,
+ * up to 64 MiB), so even a relation with a few rows reserves 64 MiB. A task
that keeps many
Review Comment:
**Nit (P3):** Only the derived default page size is capped at 64 MiB. An
explicit `spark.buffer.pageSize` has no such bound
(`MemoryManager.pageSizeBytes` uses it as is), and a small derived default can
be as low as 1 MiB, so a few-row relation used to reserve one full page of
whatever size applied, not always 64 MiB. The sibling `LongToUnsafeRowMap`
comment qualifies the same claim with "With the default page size". For example:
```suggestion
* Otherwise `BytesToBytesMap` rounds its first page up to the page size
(`spark.buffer.pageSize`
* or a default of up to 64 MiB), so even a few rows reserve a full page.
A task that keeps many
```
**Verification:**
- **Inspection:** The rationale describes the page size as the configured
spark.buffer.pageSize or a derived default of at most 64 MiB, and a few-row
relation's old reservation as one full page.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala:
##########
@@ -469,7 +481,8 @@ private[joins] object UnsafeHashedRelation {
taskMemoryManager,
// Only 70% of the slots can be used before growing, more capacity help
to reduce collision
(sizeEstimate * 1.5 + 1).toInt,
- pageSizeBytes)
+ pageSizeBytes,
Review Comment:
**Nit (P3):** Reordering the constructor's parameters, as discussed in
review, makes a swap less likely, but nothing would detect one here or in
`read`. With the two longs swapped, the map's page size would become 1 MiB and
its pages would never grow past it, so a large relation would need up to 64x
more pages and could exhaust the 8192-entry page table. The new tests would
still pass, because their relations stay within the first 1 MiB page and the
map, whose page size would show the wiring, is private to
`UnsafeHashedRelation`. A small test-visible accessor for the map's
`getPageSizeBytes()` would let the tests check that a built relation and its
Java and Kryo copies use the `SparkEnv` page size.
This extends an [existing
discussion](https://github.com/apache/spark/pull/60196#discussion_r4500475752).
**Verification:**
- **Regression:** A built UnsafeHashedRelation and its Java and Kryo
deserialized copies report the SparkEnv page size as their map's page size, and
the check fails if either call site swaps the two arguments.
--
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]
