peter-toth commented on code in PR #59639:
URL: https://github.com/apache/spark/pull/59639#discussion_r4339161593
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -920,6 +1095,35 @@ 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 can free the survivors not yet emitted, see {@link
#spillQueuedSurvivors}.
+ * 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, and a spill frees nothing. The task's memory manager keeps
every consumer until
+ * the task ends, so this lets go of its reader when the reader closes.
+ */
+ private static final class SurvivorMemory extends MemoryConsumer {
+ private volatile LateMaterializationParquetRecordReader reader;
+
+ SurvivorMemory(
+ TaskMemoryManager taskMemoryManager,
LateMaterializationParquetRecordReader reader) {
+ super(taskMemoryManager, MemoryMode.ON_HEAP);
+ this.reader = reader;
+ }
+
+ @Override
+ public long spill(long size, MemoryConsumer trigger) {
Review Comment:
Done in
[778c14954a9](https://github.com/apache/spark/pull/59639/commits/778c14954a9ede1af0fc6357b0b5546322e69a23).
`spill()` returns 0, and the key re-read with everything it needed is gone.
I'd rather not plan the spill as a follow-up for the SPJ case either:
- it frees only the queued on-heap keys, never the ranges;
- the re-read it forces allocates outside the memory manager, under the
pressure that triggered it;
- in a sorted merge every open reader also holds its current row group's
column chunks, which nothing accounts and which are usually larger than this
buffer.
With several readers per task, accounting with denial is what bounds them.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -920,6 +1095,35 @@ 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 can free the survivors not yet emitted, see {@link
#spillQueuedSurvivors}.
+ * 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, and a spill frees nothing. The task's memory manager keeps
every consumer until
+ * the task ends, so this lets go of its reader when the reader closes.
+ */
+ private static final class SurvivorMemory extends MemoryConsumer {
+ private volatile LateMaterializationParquetRecordReader reader;
+
+ SurvivorMemory(
+ TaskMemoryManager taskMemoryManager,
LateMaterializationParquetRecordReader reader) {
+ super(taskMemoryManager, MemoryMode.ON_HEAP);
Review Comment:
I'd keep the reservation off heap too:
- the ranges are on heap whatever the vectors are, and with scattered
survivors they are the larger part of the buffer;
- with the spill gone, nothing can be spilled on heap either, so the same
argument would drop the accounting everywhere.
Counting as an active task while it holds memory is what any operator that
reserves execution memory does. Its first reservation into a full pool can wait
the same way. The conf doc now says that a task holding the buffer is one of
the tasks sharing on-heap execution memory
([778c14954a9](https://github.com/apache/spark/pull/59639/commits/778c14954a9ede1af0fc6357b0b5546322e69a23)).
The release at each row group's load stays. So a row group skipped whole, or
one that cannot be read in part, holds nothing.
--
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]
