dongjoon-hyun commented on code in PR #59639:
URL: https://github.com/apache/spark/pull/59639#discussion_r4349411320
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -500,7 +551,13 @@ protected boolean loadNextRowGroup() throws IOException {
// Null means phase 1 gave the filter up, and phase 2 then reads every
row the pushed filter
// kept.
boolean filterGivenUp = survivors == null;
- if (filterGivenUp) abandonSplicing();
+ if (filterGivenUp) {
+ rangeCharge = 0L;
Review Comment:
Nit: Nothing pins this reset, the phase-1 twin of L592, which "a row group
whose filter phase 2 gives up does not count as giving splicing up" pins
through its `held`. Every test whose phase 1 gives the filter up after
reserving runs without a task memory manager, and the denials in "memory the
task's memory manager denies gives splicing up, and then the filter" give it up
before anything is reserved. Without this line, `abandonSplicing()` at L556 and
the release at L560 stop at `heldCharge()`, which still counts the ranges and
is above the reservation, so a row group whose ranges pass the cap keeps up to
the cap (64 MB by default) reserved through its plain read, until the next row
group loads. For example, `twoRowGroupsOf20` under an even-key filter at a cap
of 300 bytes reaches 300 bytes reserved (asks of 49, 49, 98 and 104), gives
splicing up at survivor 7 (back to 280) and the filter at survivor 8, with 280
still reserved; under `memoryTask(100L)` the filter goes with 98 rese
rved. Could a test read both in a memory task and expect every row unchecked,
a `held` of 0 at every batch, and give-ups of `(0, 0, 2, 0)` and `(0, 0, 0,
2)`? A helper that both give-up sites call (`rangeCharge = 0L;
abandonSplicing();` here and at L591-593) would also keep a future give-up site
from forgetting the reset.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -541,6 +599,7 @@ protected boolean loadNextRowGroup() throws IOException {
requireRowCount(dataPages, finalRowCount, blockIdx);
// A row group whose filter was given up reports nothing (see
`StorageFilterMetrics`).
Review Comment:
Nit: This comment predates the PR, but the PR made it stale: a row group
whose filter the budget gave up is now counted at L796 or L810 and logged, and
the `StorageFilterMetrics` doc it points to says "counts nothing there" only of
the first five. Relatedly, "each once" in ParquetStorageFilter.scala:51, and
"Each row group counts once" in the description, are really at most once: a row
group whose splicing the budget gave up and whose filter then goes for another
reason counts in none of the four, e.g. block 1 of the test for item 17,
outside its `(2, 0, 0, 0)`, or one with a later decoding or evaluation error in
phase 1. Could this say something like "A row group whose filter was given up
counts no saving, nor a splicing give-up", and those two places "at most once"?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -920,6 +1089,51 @@ private static void closeAll(ColumnVector[] vectors) {
}
}
+ /**
+ * The survivor buffer's account with the task's memory manager, so other
consumers see what it
+ * holds. A spill frees nothing. The buffer is bounded by {@code
maxSplicedRowGroupBytes} and
+ * drains as its batches are emitted, and freeing it would have the rest of
the row group read
+ * its key pages again, which the memory manager does not see either. A
denied reservation is
+ * handled where it is made, as passing that bound is. ON_HEAP whatever the
reader's memory mode,
+ * since off-heap vectors allocate native memory outside the Tungsten
off-heap pool. With them
+ * only the row ranges, which are always on heap, are accounted. The task's
memory manager keeps
+ * every consumer until the task ends, so a task holds one per file it
reserved for.
+ */
+ private static final class SurvivorMemory extends MemoryConsumer {
+ SurvivorMemory(TaskMemoryManager taskMemoryManager) {
+ super(taskMemoryManager, MemoryMode.ON_HEAP);
Review Comment:
Nit: Nothing pins this `ON_HEAP` for an off-heap reader. `TestMemoryManager`
grants and releases from one pool whatever the mode
(TestMemoryManager.scala:37-80), and the off-heap cases have no other consumer,
so passing the reader's `MEMORY_MODE` here should pass the whole suite. On an
executor, that would charge the ranges, which are on heap, to the off-heap
pool, which the description's "Off heap" bullet says this must not do. Where
that pool is empty, e.g. with `spark.sql.columnVector.offheap.enabled` set
while `spark.memory.offHeap.size` stays 0, every reservation would get 0, and
every row group with a survivor would give the filter up.
`ObservedTaskMemoryManager` already receives the consumer. Could it pass
`consumer.getMode` to `onAcquire`, so that the off-heap case of "the survivor
buffer is accounted to the task's execution memory" expects `ON_HEAP` for every
ask?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -912,6 +1024,63 @@ private void closeSplicingState() {
}
}
+ /** The bytes of key values buffered for splicing, which the cap weighs even
off heap. */
+ private long bufferedKeyCharge() {
+ return queuedKeyCharge + currentKeyAccumulatorCharge;
+ }
+
+ /** The accounted bytes the survivor buffer holds, see {@link
SurvivorMemory}. */
+ private long heldCharge() {
+ return MEMORY_MODE == MemoryMode.ON_HEAP ? rangeCharge +
bufferedKeyCharge() : rangeCharge;
+ }
+
+ /**
+ * Grows the reservation to cover {@code held} bytes, which the caller has
kept within the cap.
+ * It asks for at least what it already holds, up to {@link
#SURVIVOR_MEMORY_STEP_BYTES} and the
+ * cap, so a growing buffer asks only a few times and the reservation never
passes the cap. False
+ * when it is denied.
+ */
+ private boolean reserveSurvivorMemory(long held) {
+ if (survivorMemory == null) return true;
+ long missing = held - reservedBytes;
+ if (missing <= 0) return true;
+ long headroom = storageFilter.maxSplicedRowGroupBytes() - reservedBytes;
+ long ask = Math.max(missing,
+ Math.min(Math.min(reservedBytes, SURVIVOR_MEMORY_STEP_BYTES),
headroom));
+ long granted;
+ try {
+ granted = survivorMemory.acquireMemory(ask);
+ } catch (Exception | InternalError e) {
+ // Another consumer's spill and the eviction of cached blocks run
inside, so this can raise
Review Comment:
Minor: This catches `Exception | InternalError`, so any other `Error` raised
inside `acquireMemory` goes out as it is. The one a reservation can meet is the
`SparkOutOfMemoryError` (`SPILL_OUT_OF_MEMORY`) that
`TaskMemoryManager.spillConsumer` makes of another consumer's spill
`IOException`, e.g. on a full local disk (TaskMemoryManager.java:331-340). That
outcome is right: `shouldIgnoreCorruptFileException` does not take it, and the
task fails as it would from any other consumer's request. But the PR
description, which becomes the commit message, says that "a fatal error, or one
raised by a pending task kill, goes out as the reader's other error paths send
it" and that "anything else goes out wrapped in a checked `SparkException`".
This error is neither: `Executor.isFatalError` does not count it as fatal
(Executor.scala:1659), and nothing wraps it. 778c149's scaladoc said so, and as
my item 18 noted, `memoryReservationFailed` never receives it, so this catch is
the place to say it
. Could this comment add something like "An `Error` other than an
`InternalError`, such as the `SparkOutOfMemoryError` of a failed spill, goes
out as it is, since `FileScanRDD` cannot take it for a corrupt file.", and the
description's Errors bullet get a matching sub-bullet before merging?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -2119,6 +2199,270 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ // A task memory manager that calls `onAcquire` before each acquisition but
a test consumer's,
+ // with what is asked for and what the task holds then.
+ private class ObservedTaskMemoryManager(
+ memoryManager: MemoryManager,
+ onAcquire: (Long, Long) => Unit) extends
TaskMemoryManager(memoryManager, 0) {
+ override def acquireExecutionMemory(required: Long, consumer:
MemoryConsumer): Long = {
+ if (!consumer.isInstanceOf[TestMemoryConsumer]) {
+ onAcquire(required, getMemoryConsumptionForThisTask)
+ }
+ super.acquireExecutionMemory(required, consumer)
+ }
+ }
+
+ // A task whose memory manager grants at most `limit` bytes, and calls
`onAcquire` as
+ // `ObservedTaskMemoryManager` does.
+ private def memoryTask(
+ limit: Long = Long.MaxValue,
+ onAcquire: (Long, Long) => Unit = (_, _) => ()): TaskContextImpl = {
+ val memoryManager = new TestMemoryManager(spark.sparkContext.conf)
+ memoryManager.limit(limit)
+ new TaskContextImpl(stageId = 0, stageAttemptNumber = 0, partitionId = 0,
taskAttemptId = 0,
+ attemptNumber = 0, numPartitions = 1,
+ taskMemoryManager = new ObservedTaskMemoryManager(memoryManager,
onAcquire),
+ localProperties = new Properties, metricsSystem = null)
+ }
+
+ // Reads `path` under `filter` in batches of `batchSize` and a cap of `cap`,
as a task of
+ // `context`. Returns the rows, and what the task holds as each batch is
read.
+ private def readInTask(
+ path: String,
+ context: TaskContext,
+ offHeap: Boolean = false,
+ batchSize: Int = 4,
+ cap: String = "64MB",
+ filter: Expression = keyAtLeast(10L),
+ schema: StructType = checkedSchema()): (Seq[(Long, Boolean)], Seq[Long])
= {
+ var held = Seq.empty[Long]
+ val rows = readChecked(path, schema, Seq(filter), conf = Map(
+ SQLConf.PARQUET_VECTORIZED_READER_BATCH_SIZE.key -> batchSize.toString,
+ SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> offHeap.toString,
+
SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES.key -> cap),
+ keyOf = (b, i) => {
+ if (i == 0) held :+=
context.taskMemoryManager().getMemoryConsumptionForThisTask
+ b.column(0).getLong(i)
+ },
+ context = Some(context))
+ (rows, held)
+ }
+
+ // Reads `path` under `k >= 10` while another consumer holds all of the
task's memory but
+ // `margin`, and checks that the read neither drops rows nor spills that
consumer.
+ private def readBesideOther(
+ path: String,
+ rows: Long,
+ margin: Long,
+ cap: String = "64MB"): Unit = {
+ val held = 1000000L
+ val context = memoryTask(held + margin)
+ val other = new TestMemoryConsumer(context.taskMemoryManager)
+ other.use(held)
+ val (read, _) = readInTask(path, context, batchSize = 4096, cap = cap)
+ val name = s"$rows rows with a margin of $margin under a cap of $cap"
+ assert(read == marked(10L until rows, checked = true), s"$name: got
${read.size} rows")
+ assert(other.getUsed == held, s"$name: the read must not spill another
consumer")
+ }
+
+ // One row group of 400,000 rows of a long key and a string value. Under `k
>= 10` its 399,990
+ // survivors fall in one range and buffer 3,599,950 bytes, several steps of
the reservation.
+ private def oneRowGroupOf400000(dir: File): String = {
+ val path = writeSingleParquetFile(dir,
+ spark.range(400000L).selectExpr("id AS k", "CAST(id AS STRING) AS v"),
+ rowGroupSize = ParquetWriter.DEFAULT_BLOCK_SIZE)
+ assert(footerOf(path).getBlocks.size == 1, "the fixture must be one row
group")
+ path
+ }
+
+ test("the survivor buffer is accounted to the task's execution memory") {
+ // What a row group buffers is held while it is emitted, and released when
the next one loads.
+ // Phase 1 reserves a little ahead of what it buffers, and gives that back
before phase 2. Off
+ // heap, only the ranges are accounted, since the vectors' native memory
is not the memory
+ // manager's.
+ withTempDir { dir =>
+ val path = twoRowGroupsOf20(dir)
+ // Past a cap of 100 bytes, at the seventh survivor of each row group,
the keys go, a full
+ // batch of them already queued, and only the range is held.
+ Seq((false, "64MB", Seq(130L, 130L, 130L, 220L, 220L, 220L, 220L, 220L)),
+ (true, "64MB", Seq.fill(8)(40L)),
+ (false, "100b", Seq.fill(8)(40L))).foreach { case (offHeap, cap,
expected) =>
+ val context = memoryTask()
+ val (rows, held) = readInTask(path, context, offHeap, cap = cap)
+ val name = s"off heap $offHeap under a cap of $cap"
+ assert(rows == marked(10L until 40L, checked = true), s"$name: got
$rows")
+ assert(held == expected,
+ s"$name: each batch must hold its row group's reservation until the
next row group " +
+ s"loads; got $held")
+ assert(context.taskMemoryManager.getMemoryConsumptionForThisTask == 0,
+ s"$name: not released")
+ }
+ // On heap, each row group asks first for its first survivor's 49 bytes,
with nothing held
+ // since the previous one's release, and then for as much again as it
holds, three and four
+ // times for the 130 and 220 bytes it reaches.
+ var asked = Seq.empty[(Long, Long)]
+ readInTask(path, memoryTask(onAcquire = (size, held) => asked :+=
((size, held))))
+ assert(asked == Seq((49L, 0L), (49L, 49L), (98L, 98L), (49L, 0L), (49L,
49L), (98L, 98L),
+ (196L, 196L)), s"got $asked")
+ // A reader closed after its first batch, as under a limit, releases
what it still holds.
+ val context = memoryTask()
+ val file = PartitionedFile(
+ InternalRow.empty, SparkPath.fromPathString(path), 0, new
File(path).length())
+ val heldAtFirst =
withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_BATCH_SIZE.key -> "4") {
+ asTask(context) {
+ drain(formatReader(checkedSchema(), Nil,
Seq(keyAtLeast(10L)))(file)) { it =>
+ it.next()
+ context.taskMemoryManager.getMemoryConsumptionForThisTask
+ }
+ }
+ }
+ assert(heldAtFirst == 130L,
+ s"the first batch must hold the first row group; got $heldAtFirst")
+ assert(context.taskMemoryManager.getMemoryConsumptionForThisTask == 0,
+ "a reader closed early must release what it holds")
+ }
+ }
+
+ test("emitting a row group releases its memory in steps") {
+ // Each batch emitted frees its survivors, but the reservation is cut only
once it is a step
+ // past what is still buffered, so a large row group gives its memory back
a few times rather
+ // than per batch.
+ withTempDir { dir =>
+ val path = oneRowGroupOf400000(dir)
+ var asked = Seq.empty[Long]
+ val context = memoryTask(onAcquire = (size, _) => asked :+= size)
+ val (rows, reserved) = readInTask(path, context, batchSize = 4096)
+ assert(rows.size == 399990, s"got ${rows.size} rows")
+ // What batch `b` still holds, its own survivors and those after it at 9
bytes each, and the
+ // range. A batch frees 36,864 bytes, so the reservation steps down
every 29 batches.
+ def held(b: Int): Long = 40L + 9L * (399990L - 4096L * b)
+ val expected = (0 until 98).map(b => held(b - b % 29))
+ assert(reserved == expected, s"expected the reservation to step down at
batches 29, 58 " +
+ s"and 87; got ${reserved.distinct}")
+ assert(context.taskMemoryManager.getMemoryConsumptionForThisTask == 0,
"not released")
+ // A growing buffer asks a few times. First for the first survivor, then
for as much again as
+ // it holds up to 802,816 bytes, and then 1 MiB at a time.
+ assert(asked == Seq(49L) ++ (0 to 14).map(49L << _) ++ Seq.fill(2)(1L <<
20),
+ s"got ${asked.size} asks: ${asked.take(20)}")
+ // The 3,599,950 bytes it ends up buffering are not asked for twice
over. Asking for as much
+ // again as it holds would spill the other consumer.
+ readBesideOther(path, rows = 400000L, margin = 3599950L + (1L << 20))
+ }
+ }
+
+ test("memory the task's memory manager denies gives splicing up, and then
the filter") {
+ withTempDir { dir =>
+ val path = twoRowGroupsOf20(dir)
+ // Each read must leave the task holding nothing.
+ def released(context: TaskContext, available: Long): Unit =
+ assert(context.taskMemoryManager().getMemoryConsumptionForThisTask ==
0,
+ s"a read with $available bytes free must release what it reserved")
+ def rows(available: Long, offHeap: Boolean = false): Seq[(Long,
Boolean)] = {
+ val context = memoryTask(available)
+ val (read, _) = readInTask(path, context, offHeap)
+ released(context, available)
+ read
+ }
+ // The bytes a read transfers, with the row groups the budget limited,
see `budgetGiveUps`.
+ def bytes(available: Long, cap: String = "64MB"): (Long, Seq[Long]) = {
+ val metrics = metricMap()
+ val context = memoryTask(available)
+ val (_, read) = readUnderCap(path, checkedSchema(),
Seq(keyAtLeast(10L)), cap,
+ batchSize = 4, metrics, Some(context))
+ released(context, available)
+ (read, budgetGiveUps(metrics))
+ }
+ val plain = rows(Long.MaxValue)
+ assert(plain == marked(10L until 40L, checked = true), s"got $plain")
+ // 105 bytes hold the 40 of the first row group's range and 7 of its
keys, but not the 8th.
+ // So splicing is given up there, and in the second row group, as a cap
of 105 bytes gives
+ // it up. The read must cost exactly what the cap does, the key column
read again, while the
+ // filter stays. The metrics tell the two causes apart.
+ val keysDenied = rows(105L)
+ assert(keysDenied == plain, s"denied keys must keep the filter; got
$keysDenied")
+ // 50 bytes take the first survivor's 49, and then grant 1 of the 9 the
second one misses,
+ // which goes back.
+ assert(rows(50L) == plain, "a partly granted reservation must keep the
filter")
+ // 130 bytes grant the seventh survivor 32 of the 98 it asks for, which
covers the 5 it
+ // misses, so the first row group keeps splicing. The second is denied
at its eleventh.
+ val (_, partlyGranted) = bytes(130L)
+ assert(partlyGranted == Seq(0L, 1L, 0L, 0L),
+ s"only the second row group must be denied; got $partlyGranted")
+ val (uncappedBytes, uncapped) = bytes(Long.MaxValue)
+ assert(uncapped == Seq(0L, 0L, 0L, 0L), s"got $uncapped")
+ val (cappedBytes, capped) = bytes(Long.MaxValue, cap = "105b")
+ assert(cappedBytes > uncappedBytes, "a cap of 105 bytes must give
splicing up")
+ assert(capped == Seq(2L, 0L, 0L, 0L), s"both row groups pass the cap;
got $capped")
+ val ((keysDeniedBytes, denied), keysLogged) =
+ withLogged(splicingGivenUpOnMemory)(bytes(105L))
+ assert(keysDeniedBytes == cappedBytes,
+ s"denied keys must read what the cap reads; got $keysDeniedBytes
against $cappedBytes")
+ assert(denied == Seq(0L, 2L, 0L, 0L), s"both row groups are denied; got
$denied")
+ assert(keysLogged == 1, s"which is logged once per split; got
$keysLogged")
+ // Nothing at all does not hold the ranges either, so the filter is
given up. Off heap only
+ // the ranges are asked for, so 40 bytes keep the filter and nothing
gives it up.
+ assert(rows(0L) == marked(0L until 40L, checked = false),
+ "denied ranges must give the filter up")
+ val ((_, rangesDenied), rangesLogged) =
withLogged(filterGivenUpOnMemory)(bytes(0L))
+ assert(rangesDenied == Seq(0L, 0L, 0L, 2L),
+ s"both row groups count as giving the filter up, not splicing; got
$rangesDenied")
+ assert(rangesLogged == 1, s"which is logged once per split; got
$rangesLogged")
+ assert(rows(40L, offHeap = true) == plain, "off heap, ranges that fit
must keep the filter")
+ assert(rows(0L, offHeap = true) == marked(0L until 40L, checked = false),
+ "off heap, denied ranges must give the filter up")
+ }
+ }
+
+ test("an error raised while reserving fails the read in a form
ignoreCorruptFiles keeps") {
+ // Reserving the buffer can have the task's memory manager spill another
consumer, which can
+ // raise anything. A `RuntimeException`, which a kill's interrupt during a
spill also becomes,
+ // goes out checked, and an `InternalError` in a fatal `UnknownError`.
Under
+ // `ignoreCorruptFiles` either would otherwise be taken for a corrupt
file, and the rest of the
+ // file dropped.
+ withTempDir { dir =>
+ val path = twoRowGroupsOf20(dir)
+ Seq(new RuntimeException("spill"), new InternalError("spill")).foreach {
error =>
Review Comment:
Nit: These cases do not pin item 18. A `memoryReservationFailed` with only
two cases, an `InternalError` into an `UnknownError` and anything else into a
`SparkException`, should pass all three, since none of them has a fatal error
in its cause chain or runs under a pending kill, and the test checks only the
class and the cause. Could `failing` also throw something like `new
RuntimeException("spill", new OutOfMemoryError("deep"))`, and the test expect
that same `OutOfMemoryError` to come out? With two cases it would come out as
`SparkException(RuntimeException(OutOfMemoryError))`, two levels deeper, which
item 18 was about.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -337,17 +376,21 @@ private void decideAllKeysMissingFile() {
public boolean nextBatch() throws IOException {
closeAll(publishedKeyVectors);
publishedKeyVectors = null;
+ trimSurvivorMemory();
if (!super.nextBatch()) return false;
if (keyColumns != null) {
if (spliceCurrentRowGroup) {
- publishedKeyVectors = survivorBatches.pollFirst();
- if (publishedKeyVectors == null) {
+ SurvivorBatch next = survivorBatches.pollFirst();
+ if (next == null) {
// Unreachable, since the queue holds exactly the survivors the emit
loop counts. Without
// this check the key slots would show stale values.
throw ParquetStorageFilter.internalError(String.format(
"Storage-filter survivor queue of row group %d in %s ran out
with %d rows still "
+ "to emit", nextBlockIndex - 1, fileReader.getFile(),
columnarBatch.numRows()));
}
+ publishedKeyVectors = next.vectors();
+ // Still reserved, until the next batch frees these vectors and trims
the reservation.
Review Comment:
Nit: "until the next batch frees these vectors and trims the reservation"
reads as if the next batch releases this charge, but `trimSurvivorMemory()`
(L1079-1082) releases only once the reservation is 1 MiB past what is held, so
in "emitting a row group releases its memory in steps" it steps down only at
batches 29, 58 and 87. Could it say "Still reserved, until a later trim, the
next row group's load, or close() releases it."?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -202,6 +221,22 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
/** The array the emitted batch is built over, whose key slots each batch
rewrites. */
private ColumnVector[] spliceBatchColumns;
+ /** What the survivor ranges of the row group being read or emitted cost to
hold. */
+ private long rangeCharge;
+ /** The survivor buffer's account with the task's memory manager, or null
outside a task. */
Review Comment:
Nit: It is also null in a task without a memory manager (L491-496) and for a
file the filter is not applied to row by row, since
`initializeLateMaterialization` returns before L494 for those. Could it say "or
null without one, or for a file the filter is not applied to row by row", as
the `keyColumns` doc does for the latter?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -920,6 +1089,51 @@ private static void closeAll(ColumnVector[] vectors) {
}
}
+ /**
+ * The survivor buffer's account with the task's memory manager, so other
consumers see what it
Review Comment:
Nit: The charge grows per row (L895), but each set of survivor vectors is
allocated a whole batch at a time (L863-868), e.g. 4096 x 9 = 36,864 bytes for
a long key. So the set being filled, and while emitting the last partly filled
one, holds up to one batch per key column beyond the reservation; each row
group's first ask is 49 bytes. That is bounded, like the unaccounted
`keyScratchVectors`, and the conf doc already calls the charge an estimate, so
no code change is needed. Could this javadoc say so, since "so other consumers
see what it holds" reads as exact?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -118,6 +127,13 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
*/
private static final int DEFAULT_CHILD_BYTES_PER_ROW = 4;
+ /**
+ * The most the survivor buffer reserves beyond what it holds, see
+ * {@link #reserveSurvivorMemory}. The task's memory manager spills other
consumers to grant a
Review Comment:
Minor: This javadoc, and the PR description's "No spill" bullet ("Accounting
with denial already lets other consumers spill before this one is refused"),
hold only for consumers that yield to another consumer's request, and the one a
storage-filtered scan usually shares its task with does not. Storage filters
are runtime bloom filters, which `InjectRuntimeFilter` adds only for a join it
expects to shuffle (InjectRuntimeFilter.scala:432-446), so the scan's task
usually feeds a shuffle write, which past 200 partitions is an
`UnsafeShuffleWriter`. Its `ShuffleExternalSorter` returns 0 for any other
consumer's trigger (ShuffleExternalSorter.java:302), as does an
`UnsafeExternalSorter` that is still being fed
(UnsafeExternalSorter.java:227-231) or a `BytesToBytesMap` that is still being
built (BytesToBytesMap.java:1061-1064). `TaskMemoryManager` then drops it
(TaskMemoryManager.java:303), and the reservation gets only the free part of
the task's share, the case in my item 6. Only spill
able collections under on-heap Tungsten, a sorter being read, or a map with a
destructive iterator yield. The behavior itself is settled by items 6 and 19,
but since the description becomes the commit message, could it say something
like "lets other consumers that yield to another consumer's request, such as a
sorter being read, spill before this one is refused", and this javadoc "may
spill other consumers, or evict cached blocks, to grant a request in full"?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -197,18 +198,24 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
s"a cap one byte under it must give splicing up; got $underCharge
against $uncapped")
}
- // Reads `path` whole through the format under a splice cap and a batch
size, and returns the rows
- // it emitted with the bytes it transferred.
+ // Reads `path` whole through the format under a splice cap and a batch
size, as a task of
+ // `context` if one is given, and returns the rows it emitted with the bytes
it transferred.
private def readUnderCap(
path: String,
schema: StructType,
storageFilters: Seq[Expression],
maxSplicedBytes: String,
batchSize: Int,
- metrics: Map[String, SQLMetric] = metricMap()): (Int, Long) =
withSQLConf(
+ metrics: Map[String, SQLMetric] = metricMap(),
+ context: Option[TaskContext] = None): (Int, Long) = withSQLConf(
SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES.key
-> maxSplicedBytes,
SQLConf.PARQUET_VECTORIZED_READER_BATCH_SIZE.key -> batchSize.toString) {
- readCounting(path, formatReader(schema, Nil, storageFilters,
countingHadoopConf(), metrics))
+ def read(): (Int, Long) = readCounting(
+ path, formatReader(schema, Nil, storageFilters, countingHadoopConf(),
metrics))
+ context match {
Review Comment:
Nit: `readChecked` ends in the same `context match` (L2014-2017). Could both
be `context.fold(read())(asTask(_)(read()))`, or call one helper beside
`asTask` that takes an `Option[TaskContext]`?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -2048,6 +2099,35 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ test("a row group whose filter phase 2 gives up does not count as giving
splicing up") {
Review Comment:
Nit: This rebuilds the fixture and the expected rows of the test above it:
`blocks` (L2109 and L2077), the schema (L2110 and L2078), the even-key filter
(L2111-2112 and L2079-2080), the damaged file (L2113-2114 and L2089-2090), and
the expected rows (L2121-2125 against L2081 and L2095-2098). Its 2,107-byte
arithmetic and its 13, 25 and 13 batches hold only for that layout. Could the
two share a helper beside `twoRowGroupsOf20`, as item 12 did for the 40-row
fixture?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -243,9 +278,13 @@ public void close() throws IOException {
try {
closeSplicingState();
} finally {
- if (checkedVector != null) {
- checkedVector.close();
- checkedVector = null;
+ try {
+ if (checkedVector != null) {
+ checkedVector.close();
+ checkedVector = null;
+ }
+ } finally {
+ releaseSurvivorMemoryTo(0L);
Review Comment:
Nit: When `close()` runs while a spliced on-heap row group is still
reserved, `abandonSplicing()` (L1014) first releases down to the ranges and
this line then releases them: two trips through the executor-wide memory
manager lock, each ending in `notifyAll()`. That happens for a reader closed
early, as in the accounted test's early close (90, then 40 bytes), at the end
of a file in which the filter excluded no row (`nextBatch()` then returns at
VectorizedParquetRecordReader.java:404 before loading again), and after a
phase-1 error. When the filter excluded any row, L513 has already released
everything. Could `close()` zero `rangeCharge` before `closeSplicingState()`,
leaving this line as the safety net?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -727,15 +785,30 @@ private RowRanges evaluateStorageFilter(
// where a plain read fails.
reportOnce(e, SPLICING_GIVEN_UP_ON_ERROR);
abandonSplicing();
- splicedBytes = 0L;
}
}
- if (splicedBytes + rangeBytes > cap) {
+ if (bufferedKeyCharge() + rangeCharge > cap) {
// The buffer goes first, which is a no-op once splicing is given up.
+ if (spliceCurrentRowGroup) splicingGiveUp =
BudgetGiveUp.SPLICING_ON_CAP;
abandonSplicing();
- splicedBytes = 0L;
// The ranges would be thrown away, so stop.
- if (rangeBytes > cap) return null;
+ if (rangeCharge > cap) {
+ recordBudgetGiveUp(BudgetGiveUp.FILTER_ON_CAP);
+ return null;
+ }
+ }
+ // A reservation the task's memory manager denies is handled as
passing the cap is. With no
+ // key values in it, giving splicing up frees nothing, so the ranges
are denied too.
+ long held = heldCharge();
+ boolean denied = !reserveSurvivorMemory(held);
+ if (denied && held > rangeCharge) {
Review Comment:
Nit: Nothing pins this guard, or the result of the retry at L807: replacing
`held > rangeCharge` with `true`, or L807 with `denied = false`, should pass
the whole suite. The only denial without key values that the suite reaches is
the off-heap `rows(0L)`, which checks the rows only; without the guard it makes
a second 40-byte acquire, is denied again, and ends the same way. With L807
mutated, `rows(0L)` and `bytes(0L)` keep the filter past survivor 1, and
survivor 2, in the same range, is denied without key values and gives it up one
survivor later, with the same rows, the same `(0, 0, 0, 2)` and the same log
line. In production the first would repeat the acquire and its spill pass for
every denied row group off heap, which my item 6 asked to avoid, and the second
would leave a row group whose last survivor is denied keeping the filter with
its last range unreserved, counted as a splicing give-up. Could a test record
the asks of an off-heap read under `memoryTask(0L)`, expecting o
ne 40-byte ask per row group, and read under `keyAtLeast(19L)` with nothing
free, whose first row group has a single survivor, expecting `(0, 0, 0, 2)`
where the L807 mutation gives `(0, 1, 0, 1)`?
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -727,15 +785,30 @@ private RowRanges evaluateStorageFilter(
// where a plain read fails.
reportOnce(e, SPLICING_GIVEN_UP_ON_ERROR);
abandonSplicing();
- splicedBytes = 0L;
}
}
- if (splicedBytes + rangeBytes > cap) {
+ if (bufferedKeyCharge() + rangeCharge > cap) {
// The buffer goes first, which is a no-op once splicing is given up.
+ if (spliceCurrentRowGroup) splicingGiveUp =
BudgetGiveUp.SPLICING_ON_CAP;
Review Comment:
Nit: Once splicing is off, `bufferedKeyCharge()` is 0, so this block is
entered only when the ranges alone pass the cap, and it returns right after.
The unguarded `abandonSplicing()` below is then a no-op only because the
reservation never passes the cap, now that it also releases memory. Could the
two share the guard, `if (spliceCurrentRowGroup) { splicingGiveUp =
BudgetGiveUp.SPLICING_ON_CAP; abandonSplicing(); }`, as in the denial block
below, and the "no-op" clause go?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -2119,6 +2199,270 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ // A task memory manager that calls `onAcquire` before each acquisition but
a test consumer's,
+ // with what is asked for and what the task holds then.
+ private class ObservedTaskMemoryManager(
+ memoryManager: MemoryManager,
+ onAcquire: (Long, Long) => Unit) extends
TaskMemoryManager(memoryManager, 0) {
+ override def acquireExecutionMemory(required: Long, consumer:
MemoryConsumer): Long = {
+ if (!consumer.isInstanceOf[TestMemoryConsumer]) {
+ onAcquire(required, getMemoryConsumptionForThisTask)
+ }
+ super.acquireExecutionMemory(required, consumer)
+ }
+ }
+
+ // A task whose memory manager grants at most `limit` bytes, and calls
`onAcquire` as
+ // `ObservedTaskMemoryManager` does.
+ private def memoryTask(
+ limit: Long = Long.MaxValue,
+ onAcquire: (Long, Long) => Unit = (_, _) => ()): TaskContextImpl = {
+ val memoryManager = new TestMemoryManager(spark.sparkContext.conf)
+ memoryManager.limit(limit)
+ new TaskContextImpl(stageId = 0, stageAttemptNumber = 0, partitionId = 0,
taskAttemptId = 0,
+ attemptNumber = 0, numPartitions = 1,
+ taskMemoryManager = new ObservedTaskMemoryManager(memoryManager,
onAcquire),
+ localProperties = new Properties, metricsSystem = null)
+ }
+
+ // Reads `path` under `filter` in batches of `batchSize` and a cap of `cap`,
as a task of
+ // `context`. Returns the rows, and what the task holds as each batch is
read.
+ private def readInTask(
+ path: String,
+ context: TaskContext,
+ offHeap: Boolean = false,
+ batchSize: Int = 4,
+ cap: String = "64MB",
+ filter: Expression = keyAtLeast(10L),
+ schema: StructType = checkedSchema()): (Seq[(Long, Boolean)], Seq[Long])
= {
+ var held = Seq.empty[Long]
+ val rows = readChecked(path, schema, Seq(filter), conf = Map(
+ SQLConf.PARQUET_VECTORIZED_READER_BATCH_SIZE.key -> batchSize.toString,
+ SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> offHeap.toString,
+
SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES.key -> cap),
+ keyOf = (b, i) => {
+ if (i == 0) held :+=
context.taskMemoryManager().getMemoryConsumptionForThisTask
+ b.column(0).getLong(i)
+ },
+ context = Some(context))
+ (rows, held)
+ }
+
+ // Reads `path` under `k >= 10` while another consumer holds all of the
task's memory but
+ // `margin`, and checks that the read neither drops rows nor spills that
consumer.
+ private def readBesideOther(
+ path: String,
+ rows: Long,
+ margin: Long,
+ cap: String = "64MB"): Unit = {
+ val held = 1000000L
+ val context = memoryTask(held + margin)
+ val other = new TestMemoryConsumer(context.taskMemoryManager)
+ other.use(held)
+ val (read, _) = readInTask(path, context, batchSize = 4096, cap = cap)
+ val name = s"$rows rows with a margin of $margin under a cap of $cap"
+ assert(read == marked(10L until rows, checked = true), s"$name: got
${read.size} rows")
+ assert(other.getUsed == held, s"$name: the read must not spill another
consumer")
+ }
+
+ // One row group of 400,000 rows of a long key and a string value. Under `k
>= 10` its 399,990
+ // survivors fall in one range and buffer 3,599,950 bytes, several steps of
the reservation.
+ private def oneRowGroupOf400000(dir: File): String = {
+ val path = writeSingleParquetFile(dir,
+ spark.range(400000L).selectExpr("id AS k", "CAST(id AS STRING) AS v"),
+ rowGroupSize = ParquetWriter.DEFAULT_BLOCK_SIZE)
+ assert(footerOf(path).getBlocks.size == 1, "the fixture must be one row
group")
+ path
+ }
+
+ test("the survivor buffer is accounted to the task's execution memory") {
+ // What a row group buffers is held while it is emitted, and released when
the next one loads.
+ // Phase 1 reserves a little ahead of what it buffers, and gives that back
before phase 2. Off
+ // heap, only the ranges are accounted, since the vectors' native memory
is not the memory
+ // manager's.
+ withTempDir { dir =>
+ val path = twoRowGroupsOf20(dir)
+ // Past a cap of 100 bytes, at the seventh survivor of each row group,
the keys go, a full
+ // batch of them already queued, and only the range is held.
+ Seq((false, "64MB", Seq(130L, 130L, 130L, 220L, 220L, 220L, 220L, 220L)),
+ (true, "64MB", Seq.fill(8)(40L)),
+ (false, "100b", Seq.fill(8)(40L))).foreach { case (offHeap, cap,
expected) =>
+ val context = memoryTask()
+ val (rows, held) = readInTask(path, context, offHeap, cap = cap)
+ val name = s"off heap $offHeap under a cap of $cap"
+ assert(rows == marked(10L until 40L, checked = true), s"$name: got
$rows")
+ assert(held == expected,
+ s"$name: each batch must hold its row group's reservation until the
next row group " +
+ s"loads; got $held")
+ assert(context.taskMemoryManager.getMemoryConsumptionForThisTask == 0,
+ s"$name: not released")
+ }
+ // On heap, each row group asks first for its first survivor's 49 bytes,
with nothing held
+ // since the previous one's release, and then for as much again as it
holds, three and four
+ // times for the 130 and 220 bytes it reaches.
+ var asked = Seq.empty[(Long, Long)]
+ readInTask(path, memoryTask(onAcquire = (size, held) => asked :+=
((size, held))))
+ assert(asked == Seq((49L, 0L), (49L, 49L), (98L, 98L), (49L, 0L), (49L,
49L), (98L, 98L),
+ (196L, 196L)), s"got $asked")
+ // A reader closed after its first batch, as under a limit, releases
what it still holds.
+ val context = memoryTask()
+ val file = PartitionedFile(
+ InternalRow.empty, SparkPath.fromPathString(path), 0, new
File(path).length())
+ val heldAtFirst =
withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_BATCH_SIZE.key -> "4") {
+ asTask(context) {
+ drain(formatReader(checkedSchema(), Nil,
Seq(keyAtLeast(10L)))(file)) { it =>
+ it.next()
+ context.taskMemoryManager.getMemoryConsumptionForThisTask
+ }
+ }
+ }
+ assert(heldAtFirst == 130L,
+ s"the first batch must hold the first row group; got $heldAtFirst")
+ assert(context.taskMemoryManager.getMemoryConsumptionForThisTask == 0,
+ "a reader closed early must release what it holds")
+ }
+ }
+
+ test("emitting a row group releases its memory in steps") {
+ // Each batch emitted frees its survivors, but the reservation is cut only
once it is a step
+ // past what is still buffered, so a large row group gives its memory back
a few times rather
+ // than per batch.
+ withTempDir { dir =>
+ val path = oneRowGroupOf400000(dir)
+ var asked = Seq.empty[Long]
+ val context = memoryTask(onAcquire = (size, _) => asked :+= size)
+ val (rows, reserved) = readInTask(path, context, batchSize = 4096)
+ assert(rows.size == 399990, s"got ${rows.size} rows")
+ // What batch `b` still holds, its own survivors and those after it at 9
bytes each, and the
+ // range. A batch frees 36,864 bytes, so the reservation steps down
every 29 batches.
+ def held(b: Int): Long = 40L + 9L * (399990L - 4096L * b)
+ val expected = (0 until 98).map(b => held(b - b % 29))
+ assert(reserved == expected, s"expected the reservation to step down at
batches 29, 58 " +
+ s"and 87; got ${reserved.distinct}")
+ assert(context.taskMemoryManager.getMemoryConsumptionForThisTask == 0,
"not released")
+ // A growing buffer asks a few times. First for the first survivor, then
for as much again as
+ // it holds up to 802,816 bytes, and then 1 MiB at a time.
+ assert(asked == Seq(49L) ++ (0 to 14).map(49L << _) ++ Seq.fill(2)(1L <<
20),
+ s"got ${asked.size} asks: ${asked.take(20)}")
+ // The 3,599,950 bytes it ends up buffering are not asked for twice
over. Asking for as much
+ // again as it holds would spill the other consumer.
+ readBesideOther(path, rows = 400000L, margin = 3599950L + (1L << 20))
+ }
+ }
+
+ test("memory the task's memory manager denies gives splicing up, and then
the filter") {
+ withTempDir { dir =>
+ val path = twoRowGroupsOf20(dir)
+ // Each read must leave the task holding nothing.
+ def released(context: TaskContext, available: Long): Unit =
+ assert(context.taskMemoryManager().getMemoryConsumptionForThisTask ==
0,
+ s"a read with $available bytes free must release what it reserved")
+ def rows(available: Long, offHeap: Boolean = false): Seq[(Long,
Boolean)] = {
+ val context = memoryTask(available)
+ val (read, _) = readInTask(path, context, offHeap)
+ released(context, available)
+ read
+ }
+ // The bytes a read transfers, with the row groups the budget limited,
see `budgetGiveUps`.
+ def bytes(available: Long, cap: String = "64MB"): (Long, Seq[Long]) = {
+ val metrics = metricMap()
+ val context = memoryTask(available)
+ val (_, read) = readUnderCap(path, checkedSchema(),
Seq(keyAtLeast(10L)), cap,
+ batchSize = 4, metrics, Some(context))
+ released(context, available)
+ (read, budgetGiveUps(metrics))
+ }
+ val plain = rows(Long.MaxValue)
+ assert(plain == marked(10L until 40L, checked = true), s"got $plain")
+ // 105 bytes hold the 40 of the first row group's range and 7 of its
keys, but not the 8th.
+ // So splicing is given up there, and in the second row group, as a cap
of 105 bytes gives
+ // it up. The read must cost exactly what the cap does, the key column
read again, while the
+ // filter stays. The metrics tell the two causes apart.
+ val keysDenied = rows(105L)
+ assert(keysDenied == plain, s"denied keys must keep the filter; got
$keysDenied")
+ // 50 bytes take the first survivor's 49, and then grant 1 of the 9 the
second one misses,
+ // which goes back.
+ assert(rows(50L) == plain, "a partly granted reservation must keep the
filter")
+ // 130 bytes grant the seventh survivor 32 of the 98 it asks for, which
covers the 5 it
+ // misses, so the first row group keeps splicing. The second is denied
at its eleventh.
+ val (_, partlyGranted) = bytes(130L)
+ assert(partlyGranted == Seq(0L, 1L, 0L, 0L),
+ s"only the second row group must be denied; got $partlyGranted")
+ val (uncappedBytes, uncapped) = bytes(Long.MaxValue)
+ assert(uncapped == Seq(0L, 0L, 0L, 0L), s"got $uncapped")
+ val (cappedBytes, capped) = bytes(Long.MaxValue, cap = "105b")
+ assert(cappedBytes > uncappedBytes, "a cap of 105 bytes must give
splicing up")
+ assert(capped == Seq(2L, 0L, 0L, 0L), s"both row groups pass the cap;
got $capped")
+ val ((keysDeniedBytes, denied), keysLogged) =
+ withLogged(splicingGivenUpOnMemory)(bytes(105L))
+ assert(keysDeniedBytes == cappedBytes,
+ s"denied keys must read what the cap reads; got $keysDeniedBytes
against $cappedBytes")
+ assert(denied == Seq(0L, 2L, 0L, 0L), s"both row groups are denied; got
$denied")
+ assert(keysLogged == 1, s"which is logged once per split; got
$keysLogged")
+ // Nothing at all does not hold the ranges either, so the filter is
given up. Off heap only
+ // the ranges are asked for, so 40 bytes keep the filter and nothing
gives it up.
+ assert(rows(0L) == marked(0L until 40L, checked = false),
+ "denied ranges must give the filter up")
+ val ((_, rangesDenied), rangesLogged) =
withLogged(filterGivenUpOnMemory)(bytes(0L))
+ assert(rangesDenied == Seq(0L, 0L, 0L, 2L),
+ s"both row groups count as giving the filter up, not splicing; got
$rangesDenied")
+ assert(rangesLogged == 1, s"which is logged once per split; got
$rangesLogged")
+ assert(rows(40L, offHeap = true) == plain, "off heap, ranges that fit
must keep the filter")
+ assert(rows(0L, offHeap = true) == marked(0L until 40L, checked = false),
+ "off heap, denied ranges must give the filter up")
+ }
+ }
+
+ test("an error raised while reserving fails the read in a form
ignoreCorruptFiles keeps") {
+ // Reserving the buffer can have the task's memory manager spill another
consumer, which can
+ // raise anything. A `RuntimeException`, which a kill's interrupt during a
spill also becomes,
+ // goes out checked, and an `InternalError` in a fatal `UnknownError`.
Under
+ // `ignoreCorruptFiles` either would otherwise be taken for a corrupt
file, and the rest of the
+ // file dropped.
+ withTempDir { dir =>
+ val path = twoRowGroupsOf20(dir)
+ Seq(new RuntimeException("spill"), new InternalError("spill")).foreach {
error =>
+ val context = memoryTask(1000L)
+ val failing = new TestMemoryConsumer(context.taskMemoryManager) {
+ override def spill(size: Long, trigger: MemoryConsumer): Long =
throw error
+ }
+ failing.use(1000L)
+ val thrown = intercept[Throwable](readInTask(path, context))
+ val expected = if (error.isInstanceOf[InternalError]) {
+ classOf[UnknownError]
+ } else {
+ classOf[SparkException]
+ }
+ assert(thrown.getClass == expected && (thrown.getCause eq error),
+ s"expected $error inside ${expected.getSimpleName}; got $thrown")
+ assert(!DataSourceUtils.shouldIgnoreCorruptFileException(thrown),
s"got $thrown")
+ }
+ // Evicting a cached block to disk can raise an `IOException` there too,
which the task's
+ // memory manager passes on as it is.
+ val full = new IOException("No space left on device")
+ val evicting = memoryTask(onAcquire = (_, _) => throw full)
+ val thrown = intercept[Throwable](readInTask(path, evicting))
+ assert(thrown.getClass == classOf[SparkException] && (thrown.getCause eq
full),
+ s"expected $full inside a SparkException; got $thrown")
+ assert(!DataSourceUtils.shouldIgnoreCorruptFileException(thrown), s"got
$thrown")
+ }
+ }
+
+ test("the survivor buffer asks for little more memory than it holds") {
+ // Another consumer holds all but what the buffer needs plus a margin. The
task's memory manager
+ // spills that consumer to grant a reservation past what is free, so the
buffer must not ask
+ // for more than its margin: less than as much again as it holds while
small, less than 1 MiB
+ // more once it is past that, see "emitting a row group releases its
memory in steps", and
+ // never past the cap.
+ withTempDir { dir =>
+ // The buffer never holds more than 220 bytes, so it asks for less than
440.
+ val path = twoRowGroupsOf20(dir)
+ readBesideOther(path, rows = 40L, margin = 440L)
+ // Doubling from 196 would ask for 392, which a cap of 300 bytes stops
at 300.
Review Comment:
Nit: At survivor 18 of the second row group the buffer holds 202 bytes with
196 reserved, so it asks for 196, reaching 392, at the default cap, and for
104, reaching 300, at 300 bytes; the accounted test records that ask as `(196L,
196L)` at L2306. So "would ask for 392" mixes the ask with the reservation.
Could it say "Doubling from 196 would reserve 392, which a cap of 300 bytes
stops at 300."?
--
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]
