peter-toth commented on code in PR #59639:
URL: https://github.com/apache/spark/pull/59639#discussion_r4339166584
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -183,20 +188,44 @@ public class LateMaterializationParquetRecordReader
extends VectorizedParquetRec
private WritableColumnVector checkedVector;
/**
- * Splicing state. {@link #survivorBatches} holds one array of survivor
vectors per batch to emit,
- * indexed like {@link #keyColumns}, and {@link #currentKeyAccumulators} the
array still filling.
- * Together they hold every survivor of one row group, outside any
MemoryConsumer.
+ * Splicing state. {@link #survivorBatches} holds one {@link SurvivorBatch}
per batch to emit, and
+ * {@link #currentKeyAccumulators} the vectors still filling, indexed like
{@link #keyColumns}.
+ * Together they hold the survivors of one row group not yet emitted.
*/
- private final ArrayDeque<WritableColumnVector[]> survivorBatches = new
ArrayDeque<>();
+ private final ArrayDeque<SurvivorBatch> survivorBatches = new ArrayDeque<>();
/** The survivor vectors still filling, or null when there are none. */
private WritableColumnVector[] currentKeyAccumulators;
- /** Row count of {@link #currentKeyAccumulators}. */
+ /** Row count and charge of {@link #currentKeyAccumulators}. */
private int currentKeyAccumulatorRowCount;
+ private long currentKeyAccumulatorCharge;
/**
- * The survivor vectors the current batch is built on. The next batch frees
them, or
- * {@link #close()} when no next batch comes.
+ * The survivor vectors the current batch is built on, and their charge. The
next batch frees
+ * them, or {@link #close()} when no next batch comes.
*/
private WritableColumnVector[] publishedKeyVectors;
+ private long publishedKeyCharge;
+ /** 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. */
+ private SurvivorMemory survivorMemory;
+ /** Whether the key values are accounted too, which needs them on heap. */
+ private boolean accountKeys;
+ /** The thread that reads, the only one whose spill may touch this reader's
state. */
+ private volatile Thread readerThread;
+
+ /** The block, final ranges and row count of the row group being emitted. */
+ private int loadedBlockIndex;
+ private RowRanges loadedRanges;
+ private long loadedRowCount;
+ /** Whether that row group can be spilled, which {@link #loadNextRowGroup}
decides. */
+ private boolean spillableRowGroup;
+ /** The rows of that row group already emitted. */
+ private long rowsEmittedInRowGroup;
+ /** Set by a spill, so the next batch reads the keys of the rest of the row
group again. */
+ private boolean rereadRemainingKeys;
+ /** The key pages read again after a spill, closed with the row group. */
+ private PageReadStore rereadKeyPages;
Review Comment:
Done in
[778c14954a9](https://github.com/apache/spark/pull/59639/commits/778c14954a9ede1af0fc6357b0b5546322e69a23).
--
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]
