peter-toth commented on code in PR #58895:
URL: https://github.com/apache/spark/pull/58895#discussion_r4084289903
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/SpecificParquetRecordReaderBase.java:
##########
@@ -87,6 +87,13 @@ public abstract class SpecificParquetRecordReaderBase<T>
extends RecordReader<Vo
protected ParquetRowGroupReader reader;
+ /**
+ * The opened input file and parquet footer. Stored so subclasses can read
footer-derived metadata
+ * without re-opening the file. Set by both {@link #initialize} overloads.
+ */
+ protected HadoopInputFile inputFile;
Review Comment:
Dropped, both of them, and the footer is a local again. The javadoc went
with them.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +929,450 @@ private void checkEndOfRowGroup() throws IOException {
totalCountLoadedSoFar += pages.getRowCount();
}
+ /**
+ * Loads the next row group using the three-phase late-materialization
pattern, all driven by the
+ * single {@link #lateMatReader} with its requested schema mutated per phase:
+ * - Phase 0 (full schema): compute {@code pushedFilterRanges} from the
pushed data filter via
+ * column index (metadata-only) using {@link
ParquetFileReader#getRowRanges}.
+ * - Phase 1 (key-only schema): read key-column pages restricted to {@code
pushedFilterRanges},
+ * evaluate the storage filter per row, build {@code finalRanges}.
+ * - Phase 2 (non-key schema): read non-key columns restricted to {@code
finalRanges}.
+ * Skipped entirely when {@link #nonKeyRequestedSchema} is null
(all-keys projection).
+ *
+ * Row groups for which {@code finalRanges} is empty are skipped entirely
(no phase-2 IO).
+ * Sets {@link #hitEndOfData} when all row groups have been processed.
+ */
+ private void loadNextRowGroupWithLateMaterialization() throws IOException {
+ while (nextBlockIndex < totalBlockCount) {
+ int blockIdx = nextBlockIndex++;
+ long blockRowCount =
lateMatReader.getRowGroups().get(blockIdx).getRowCount();
+ if (blockRowCount == 0) {
+ // parquet-mr never writes these, but RowRanges.createSingle(0) would
build Range(0, -1) and
+ // trip parquet's own `from <= to` assertion. The plain read path
skips them too.
+ continue;
+ }
+
+ // Phase 0: rows allowed by the pushed data filter (column-index
granularity). Restore the
+ // full requestedSchema first: phases 1 and 2 below narrow the reader's
schema, and
+ // ParquetFileReader.getRowRanges computes ranges against whatever
schema is set (it passes
+ // the reader's current `paths` to ColumnIndexFilter). This is defensive
-- with column-index
+ // filtering on, getFilteredRecordCount() in initialize() has already
memoized every block's
+ // ranges under the full schema, and with it off we do not call
getRowRanges at all.
+ lateMatReader.setRequestedSchema(requestedSchema);
+ // ParquetFileReader.getRowRanges only checks whether a filter is
pushed, NOT
+ // options.useColumnIndexFilter(), so calling it unconditionally would
keep applying
+ // column-index filtering after a user turned it off. That conf is the
documented escape hatch
+ // for files whose column index is wrong, and trusting a wrong column
index here would drop
+ // rows for good: finalRanges is a subset of pushedFilterRanges, and the
post-scan Filter no
+ // longer holds this predicate.
+ RowRanges pushedFilterRanges = useColumnIndexFilter
+ ? lateMatReader.getRowRanges(blockIdx)
+ : RowRanges.createSingle(blockRowCount);
+ if (pushedFilterRanges.rowCount() == 0) {
+ // Pushed data filter rejects this block entirely via column index.
Not a storage-filter
+ // skip, so we don't increment storage-filter metrics.
+ continue;
+ }
+
+ // Both byte metrics are always wired in production (FileSourceScanLike
creates all five
+ // whenever storageFilters is non-empty), so this only skips the work on
the test-only path
+ // that drives the reader directly. compressedBytesForRowRanges never
does IO of its own, so
+ // there is nothing here to avoid on the production path.
+ StorageFilterMetrics m = storageFilter.metrics();
+ SQLMetric bytesAvoidedRg = m.bytesAvoidedByRowGroup();
+ SQLMetric bytesAvoidedPf = m.bytesAvoidedByPageFiltering();
+ boolean needBytes = bytesAvoidedRg != null || bytesAvoidedPf != null;
+ long baselineRows = pushedFilterRanges.rowCount();
+ long baselineBytes = needBytes
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx,
requestedSchema,
+ pushedFilterRanges)
+ : 0L;
+
+ // Phase 1: switch to key-only schema, read key columns under
pushedFilterRanges, evaluate the
+ // storage filter per row.
+ lateMatReader.setRequestedSchema(keyOnlyRequestedSchema);
+ long phase1Bytes = needBytes
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx,
keyOnlyRequestedSchema,
+ pushedFilterRanges)
+ : 0L;
+ PageReadStore keyPages = lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges);
+ if (keyPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block, and we already
+ // know pushedFilterRanges selects at least one row. Skipping the
block here would drop its
+ // surviving rows from the output, so assert rather than `continue`.
+ throw new IllegalStateException(
+ "No key pages for row group " + blockIdx + " despite "
+ + pushedFilterRanges.rowCount() + " rows selected by the
pushed filter");
+ }
+ RowRanges finalRanges = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+
+ if (finalRanges.rowCount() == 0) {
+ // Every surviving row was rejected by the storage filter; skip block
entirely. We still
+ // paid phase-1 to read the key column, so the bytes avoided vs a
no-storage-filter read
+ // are baseline - phase1 (the non-key bytes the no-filter path would
have read).
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcludedRg = m.rowsExcludedByRowGroup();
+ if (rowsExcludedRg != null) rowsExcludedRg.add(baselineRows);
+ if (bytesAvoidedRg != null) bytesAvoidedRg.add(baselineBytes -
phase1Bytes);
+ continue;
+ }
+
+ // Phase 2: switch to non-key schema, read non-key columns under
finalRanges. Skipped entirely
+ // when the projection is all keys (nonKeyRequestedSchema is null); emit
reconstructs each
+ // batch from the key queues alone.
+ long keptRows;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ if (nonKeyRequestedSchema == null) {
+ keptRows = finalRanges.rowCount();
+ phase2Bytes = 0L;
+ } else {
+ lateMatReader.setRequestedSchema(nonKeyRequestedSchema);
+ // requireOffsetIndexesForPhase2() already established that every
projected column of every
+ // row group has an offset index, so this page-filtering read cannot
fail for want of one.
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx, finalRanges);
+ if (dataPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block or empty ranges,
+ // both excluded above. Match phase 1 and fail with a message rather
than an NPE.
+ throw new IllegalStateException(
+ "No data pages for row group " + blockIdx + " despite " +
finalRanges.rowCount()
+ + " surviving rows");
+ }
+ keptRows = dataPages.getRowCount();
+ phase2Bytes = bytesAvoidedPf != null
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx,
nonKeyRequestedSchema,
+ finalRanges)
+ : 0L;
+ }
+ long filteredRows = baselineRows - keptRows;
+ SQLMetric rowsExcludedWithinRg = m.rowsExcludedWithinRowGroup();
+ if (rowsExcludedWithinRg != null && filteredRows > 0)
rowsExcludedWithinRg.add(filteredRows);
+ if (bytesAvoidedPf != null) {
+ bytesAvoidedPf.add(baselineBytes - phase1Bytes - phase2Bytes);
+ }
+
+ if (dataPages != null) {
+ if (rowIndexGenerator != null) {
+ rowIndexGenerator.initFromPageReadStore(dataPages);
+ }
+ for (int i = 0; i < columnVectors.length; i++) {
+ if (isKeyTopLevel[i]) {
+ // Key columns are sourced from the queues during emit; skip
phase-2 reader init.
+ continue;
+ }
+ initColumnReader(dataPages, columnVectors[i]);
+ }
+ }
+ totalCountLoadedSoFar += keptRows;
+ return;
+ }
+ hitEndOfData = true;
+ }
+
+ /**
+ * Compressed bytes the reader transfers for the leaf columns of {@code
schema} when it reads
+ * exactly {@code rowRanges} of the given block. Page headers and the
dictionary page are
+ * included, because both are read whenever any page of a chunk is read.
+ *
+ * <p>Two sources, chosen so this never causes IO of its own:
+ * <ul>
+ * <li>{@code rowRanges} covers the whole block: the answer is the sum of
the chunks'
+ * {@code getTotalSize()}, which is already in the footer. This is the
case that matters --
+ * whenever nothing else has built the block's {@link
ColumnIndexStore}, {@code rowRanges}
+ * is necessarily the whole block, because a narrower range can only
come from column-index
+ * filtering, which builds the store as a side effect.
+ * <li>{@code rowRanges} is a strict subset: walk the offset index, as
parquet's own read path
+ * does, and add the dictionary page the way {@code
calculateOffsetRanges} does. The store
+ * is guaranteed to exist here, so the walk is pure metadata
arithmetic.
+ * </ul>
+ *
+ * <p>Columns absent from this physical file (schema evolution) contribute
nothing, which is
+ * correct: the reader transfers nothing for them. Every caller for a given
block walks the same
+ * metadata, so a skipped column drops out of the baseline and the per-phase
totals alike.
+ */
+ private static long compressedBytesForRowRanges(
+ ParquetFileReader reader,
+ int blockIndex,
+ MessageType schema,
+ RowRanges rowRanges) {
+ if (schema == null || rowRanges.rowCount() == 0 ||
schema.getColumns().isEmpty()) {
+ return 0L;
+ }
+ BlockMetaData block = reader.getRowGroups().get(blockIndex);
+ long blockRowCount = block.getRowCount();
+ Map<ColumnPath, ColumnChunkMetaData> chunks = new HashMap<>();
+ for (ColumnChunkMetaData chunk : block.getColumns()) {
+ chunks.put(chunk.getPath(), chunk);
+ }
+ boolean wholeBlock = rowRanges.rowCount() == blockRowCount;
+ ColumnIndexStore ciStore = wholeBlock ? null :
reader.getColumnIndexStore(blockIndex);
+ long total = 0L;
+ for (ColumnDescriptor column : schema.getColumns()) {
+ ColumnPath path = ColumnPath.get(column.getPath());
+ ColumnChunkMetaData chunk = chunks.get(path);
+ if (chunk == null) {
+ // Column is in the (clipped) requested schema but not in this file.
+ continue;
+ }
+ if (wholeBlock) {
+ total += chunk.getTotalSize();
+ continue;
+ }
+ OffsetIndex offsetIndex;
+ try {
+ offsetIndex = ciStore.getOffsetIndex(path);
+ } catch (MissingOffsetIndexException e) {
+ continue;
+ }
+ if (offsetIndex == null) {
+ continue;
+ }
+ // The dictionary page is read whenever any data page of the chunk is,
so count it here the
+ // same way parquet's ColumnIndexFilterUtils.calculateOffsetRanges does.
+ total += dictionaryPageSize(chunk);
+ int pageCount = offsetIndex.getPageCount();
+ for (int i = 0; i < pageCount; i++) {
+ long from = offsetIndex.getFirstRowIndex(i);
+ long to = offsetIndex.getLastRowIndex(i, blockRowCount);
+ if (rowRanges.isOverlapping(from, to)) {
+ total += offsetIndex.getCompressedPageSize(i);
+ }
+ }
+ }
+ return total;
+ }
+
+ /**
+ * Compressed size of a chunk's dictionary page, or 0 if it has none.
+ * {@link ColumnChunkMetaData#getStartingPos()} already resolves to the
dictionary page offset
+ * when there is a valid one, so the gap up to the first data page is
exactly the dictionary page.
+ */
+ private static long dictionaryPageSize(ColumnChunkMetaData chunk) {
+ long startingPos = chunk.getStartingPos();
+ long firstDataPageOffset = chunk.getFirstDataPageOffset();
+ return startingPos < firstDataPageOffset ? firstDataPageOffset -
startingPos : 0L;
+ }
+
+ /**
+ * Reads all rows of the given key-only {@link PageReadStore} (which
contains only rows in
+ * {@code pushedFilterRanges}) in capacity-sized chunks, evaluates the
storage filter on each row,
+ * builds a {@link RowRanges} of surviving rows in original block-row
coordinates, and appends
+ * survivor key values into the per-key-column accumulators ({@link
#currentKeyAccumulators}).
+ * When an accumulator hits {@link #capacity}, it's pushed into {@link
#keyVectorQueues} and a
+ * fresh one is allocated. After all rows have been examined, any partial
trailing accumulator is
+ * pushed too.
+ *
+ * <p>The result is a subset of {@code pushedFilterRanges}: rows not in
{@code
+ * pushedFilterRanges} were never read and are implicitly excluded.
+ */
+ private RowRanges evaluateStorageFilter(
+ PageReadStore keyPages,
+ RowRanges pushedFilterRanges) throws IOException {
+ ensureKeyScratchAllocated();
+ VectorizedColumnReader[] readers = new
VectorizedColumnReader[keyDescriptors.length];
+ for (int i = 0; i < readers.length; i++) {
+ readers[i] = new VectorizedColumnReader(
+ keyDescriptors[i], keyRequired[i], keyPages, convertTz,
datetimeRebaseMode,
+ datetimeRebaseTz, int96RebaseMode, int96RebaseTz, writerVersion);
+ }
+ ensureCurrentKeyAccumulatorsAllocated();
+
+ long keyRowsTotal = pushedFilterRanges.rowCount();
+ PrimitiveIterator.OfLong rowIndexIter = pushedFilterRanges.iterator();
+ RowRanges.Builder finalRangesBuilder = RowRanges.builder();
+ long remaining = keyRowsTotal;
+ while (remaining > 0) {
+ int num = (int) Math.min((long) capacity, remaining);
+ for (int i = 0; i < keyScratchVectors.length; i++) {
+ keyScratchVectors[i].reset();
+ readers[i].readBatch(num, keyScratchVectors[i], null, null);
+ }
+ keyScratchBatch.setNumRows(num);
+ for (int r = 0; r < num; r++) {
+ long blockRow = rowIndexIter.nextLong();
+ if (storageFilter.test(keyScratchBatch.getRow(r))) {
+ finalRangesBuilder.addSelectedRow(blockRow);
+ appendSurvivorRowToAccumulators(r);
+ }
+ }
+ remaining -= num;
+ }
+
+ finalizePartialAccumulators();
+
+ return finalRangesBuilder.build();
+ }
+
+ private void ensureKeyScratchAllocated() {
+ if (keyScratchVectors != null) return;
+ keyScratchVectors = new WritableColumnVector[keyDescriptors.length];
+ boolean useOffHeap = MEMORY_MODE == MemoryMode.OFF_HEAP;
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ for (int i = 0; i < keyDescriptors.length; i++) {
+ DataType dt = sparkRequestedSchema.fields()[keyIndices[i]].dataType();
+ keyScratchVectors[i] = useOffHeap
+ ? new OffHeapColumnVector(capacity, dt)
+ : new OnHeapColumnVector(capacity, dt);
+ }
+ keyScratchBatch = new ColumnarBatch(keyScratchVectors);
+ }
+
+ /**
+ * Allocates the per-key-column accumulator vectors if any slot is null
(i.e. the previous
+ * accumulator was just pushed to the queue or this is the first row group).
Each accumulator has
+ * {@link #capacity} rows.
+ */
+ private void ensureCurrentKeyAccumulatorsAllocated() {
+ boolean useOffHeap = MEMORY_MODE == MemoryMode.OFF_HEAP;
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ for (int i = 0; i < currentKeyAccumulators.length; i++) {
+ if (currentKeyAccumulators[i] == null) {
+ DataType dt = sparkRequestedSchema.fields()[keyIndices[i]].dataType();
+ currentKeyAccumulators[i] = useOffHeap
+ ? new OffHeapColumnVector(capacity, dt)
+ : new OnHeapColumnVector(capacity, dt);
+ }
+ }
+ currentKeyAccumulatorRowCount = 0;
+ }
+
+ /**
+ * Appends row {@code srcRow} of each {@link #keyScratchVectors} into the
corresponding
+ * {@link #currentKeyAccumulators}. When the accumulators fill, they're
pushed onto their queues
+ * and fresh ones allocated. All key columns are appended in lockstep so
accumulators stay
+ * aligned.
+ */
+ private void appendSurvivorRowToAccumulators(int srcRow) {
+ final int dstRow = currentKeyAccumulatorRowCount;
+ final WritableColumnVector[] accs = currentKeyAccumulators;
+ final WritableColumnVector[] srcs = keyScratchVectors;
+ final ValueCopier[] copiers = keyCopiers;
+ for (int i = 0, n = accs.length; i < n; i++) {
+ WritableColumnVector src = srcs[i];
+ WritableColumnVector dst = accs[i];
+ if (src.isNullAt(srcRow)) {
+ dst.putNull(dstRow);
+ } else {
+ copiers[i].copy(dst, dstRow, src, srcRow);
+ }
+ }
+ currentKeyAccumulatorRowCount = dstRow + 1;
+ if (currentKeyAccumulatorRowCount == capacity) {
+ for (int i = 0; i < currentKeyAccumulators.length; i++) {
+ keyVectorQueues[i].addLast(currentKeyAccumulators[i]);
+ currentKeyAccumulators[i] = null;
+ }
+ ensureCurrentKeyAccumulatorsAllocated();
+ }
+ }
+
+ /**
+ * Pushes any partially-filled accumulator into its queue at row-group end
so the emit path can
+ * dequeue it as the row group's final batch.
+ */
+ private void finalizePartialAccumulators() {
+ if (currentKeyAccumulatorRowCount == 0) return;
+ for (int i = 0; i < currentKeyAccumulators.length; i++) {
+ keyVectorQueues[i].addLast(currentKeyAccumulators[i]);
+ currentKeyAccumulators[i] = null;
+ }
+ currentKeyAccumulatorRowCount = 0;
+ }
+
+ /**
+ * Closes anything held by the splicing path: pending dequeued key vectors
not yet rolled over,
+ * any vectors still queued (e.g. on early termination), and
partially-filled accumulators.
+ * Called from {@link #close()}.
+ */
+ private void closeSplicingState() {
+ if (pendingCloseKeyVectors != null) {
+ for (WritableColumnVector v : pendingCloseKeyVectors) {
+ if (v != null) v.close();
+ }
+ pendingCloseKeyVectors = null;
+ }
+ if (keyVectorQueues != null) {
+ for (java.util.ArrayDeque<WritableColumnVector> q : keyVectorQueues) {
+ if (q != null) {
+ for (WritableColumnVector v : q) v.close();
+ q.clear();
+ }
+ }
+ }
+ if (currentKeyAccumulators != null) {
+ for (WritableColumnVector v : currentKeyAccumulators) {
+ if (v != null) v.close();
+ }
+ currentKeyAccumulators = null;
+ }
+ }
+
+ /**
+ * Per-key-column value copier: appends one value from {@code src[srcRow]} to
+ * {@code dst[dstRow]}. Picked once at init via {@link
#copierFor(DataType)}; called per surviving
+ * row in {@link #appendSurvivorRowToAccumulators}. Caller handles null
sources.
+ */
+ @FunctionalInterface
+ private interface ValueCopier {
+ void copy(WritableColumnVector dst, int dstRow, WritableColumnVector src,
int srcRow);
+ }
+
+ /**
+ * Returns a {@link ValueCopier} for the given key {@link DataType}. The set
of types handled here
+ * is the definition behind {@code ParquetStorageFilter.isSupportedKeyType},
which gates both
+ * planning-time extraction and {@code ParquetStorageFilter.create} -- so
the throw at the end is
+ * unreachable. Teach both sides at once when adding a type; a type admitted
there but missing
+ * here becomes a task failure instead of a planning-time rejection.
+ */
+ private static ValueCopier copierFor(DataType dt) {
+ if (dt instanceof BooleanType) {
+ return (dst, dRow, src, sRow) -> dst.putBoolean(dRow,
src.getBoolean(sRow));
+ }
+ if (dt instanceof ByteType) {
+ return (dst, dRow, src, sRow) -> dst.putByte(dRow, src.getByte(sRow));
+ }
+ if (dt instanceof ShortType) {
+ return (dst, dRow, src, sRow) -> dst.putShort(dRow, src.getShort(sRow));
+ }
+ if (dt instanceof IntegerType
+ || dt instanceof DateType
+ || dt instanceof YearMonthIntervalType) {
+ return (dst, dRow, src, sRow) -> dst.putInt(dRow, src.getInt(sRow));
+ }
+ if (dt instanceof LongType
+ || dt instanceof TimestampType
+ || dt instanceof TimestampNTZType
+ || dt instanceof TimeType
+ || dt instanceof DayTimeIntervalType) {
+ return (dst, dRow, src, sRow) -> dst.putLong(dRow, src.getLong(sRow));
+ }
+ if (dt instanceof FloatType) {
+ return (dst, dRow, src, sRow) -> dst.putFloat(dRow, src.getFloat(sRow));
+ }
+ if (dt instanceof DoubleType) {
+ return (dst, dRow, src, sRow) -> dst.putDouble(dRow,
src.getDouble(sRow));
+ }
+ if (dt instanceof DecimalType decimalType) {
+ int precision = decimalType.precision();
+ if (precision <= Decimal.MAX_INT_DIGITS()) {
+ return (dst, dRow, src, sRow) -> dst.putInt(dRow, src.getInt(sRow));
+ }
+ if (precision <= Decimal.MAX_LONG_DIGITS()) {
+ return (dst, dRow, src, sRow) -> dst.putLong(dRow, src.getLong(sRow));
+ }
+ return (dst, dRow, src, sRow) -> dst.putByteArray(dRow,
src.getBinary(sRow));
+ }
+ if (dt instanceof StringType
+ || dt instanceof VarcharType
+ || dt instanceof CharType
+ || dt instanceof BinaryType) {
+ return (dst, dRow, src, sRow) -> dst.putByteArray(dRow,
src.getBinary(sRow));
Review Comment:
Real, and left as a follow-up named in the description. A vector-to-vector
`appendBytes` means touching `WritableColumnVector` and both subclasses, which
widens this change into the core column-vector classes. It is also confined to
string and binary keys -- a fixed-width copier writes straight into the
destination array with no allocation -- so the cost is not on every shape.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +929,450 @@ private void checkEndOfRowGroup() throws IOException {
totalCountLoadedSoFar += pages.getRowCount();
}
+ /**
+ * Loads the next row group using the three-phase late-materialization
pattern, all driven by the
+ * single {@link #lateMatReader} with its requested schema mutated per phase:
+ * - Phase 0 (full schema): compute {@code pushedFilterRanges} from the
pushed data filter via
+ * column index (metadata-only) using {@link
ParquetFileReader#getRowRanges}.
+ * - Phase 1 (key-only schema): read key-column pages restricted to {@code
pushedFilterRanges},
+ * evaluate the storage filter per row, build {@code finalRanges}.
+ * - Phase 2 (non-key schema): read non-key columns restricted to {@code
finalRanges}.
+ * Skipped entirely when {@link #nonKeyRequestedSchema} is null
(all-keys projection).
+ *
+ * Row groups for which {@code finalRanges} is empty are skipped entirely
(no phase-2 IO).
+ * Sets {@link #hitEndOfData} when all row groups have been processed.
+ */
+ private void loadNextRowGroupWithLateMaterialization() throws IOException {
+ while (nextBlockIndex < totalBlockCount) {
+ int blockIdx = nextBlockIndex++;
+ long blockRowCount =
lateMatReader.getRowGroups().get(blockIdx).getRowCount();
+ if (blockRowCount == 0) {
+ // parquet-mr never writes these, but RowRanges.createSingle(0) would
build Range(0, -1) and
+ // trip parquet's own `from <= to` assertion. The plain read path
skips them too.
+ continue;
+ }
+
+ // Phase 0: rows allowed by the pushed data filter (column-index
granularity). Restore the
+ // full requestedSchema first: phases 1 and 2 below narrow the reader's
schema, and
+ // ParquetFileReader.getRowRanges computes ranges against whatever
schema is set (it passes
+ // the reader's current `paths` to ColumnIndexFilter). This is defensive
-- with column-index
+ // filtering on, getFilteredRecordCount() in initialize() has already
memoized every block's
+ // ranges under the full schema, and with it off we do not call
getRowRanges at all.
+ lateMatReader.setRequestedSchema(requestedSchema);
+ // ParquetFileReader.getRowRanges only checks whether a filter is
pushed, NOT
+ // options.useColumnIndexFilter(), so calling it unconditionally would
keep applying
+ // column-index filtering after a user turned it off. That conf is the
documented escape hatch
+ // for files whose column index is wrong, and trusting a wrong column
index here would drop
+ // rows for good: finalRanges is a subset of pushedFilterRanges, and the
post-scan Filter no
+ // longer holds this predicate.
+ RowRanges pushedFilterRanges = useColumnIndexFilter
+ ? lateMatReader.getRowRanges(blockIdx)
+ : RowRanges.createSingle(blockRowCount);
+ if (pushedFilterRanges.rowCount() == 0) {
+ // Pushed data filter rejects this block entirely via column index.
Not a storage-filter
+ // skip, so we don't increment storage-filter metrics.
+ continue;
+ }
+
+ // Both byte metrics are always wired in production (FileSourceScanLike
creates all five
+ // whenever storageFilters is non-empty), so this only skips the work on
the test-only path
+ // that drives the reader directly. compressedBytesForRowRanges never
does IO of its own, so
+ // there is nothing here to avoid on the production path.
+ StorageFilterMetrics m = storageFilter.metrics();
+ SQLMetric bytesAvoidedRg = m.bytesAvoidedByRowGroup();
+ SQLMetric bytesAvoidedPf = m.bytesAvoidedByPageFiltering();
+ boolean needBytes = bytesAvoidedRg != null || bytesAvoidedPf != null;
+ long baselineRows = pushedFilterRanges.rowCount();
+ long baselineBytes = needBytes
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx,
requestedSchema,
+ pushedFilterRanges)
+ : 0L;
+
+ // Phase 1: switch to key-only schema, read key columns under
pushedFilterRanges, evaluate the
+ // storage filter per row.
+ lateMatReader.setRequestedSchema(keyOnlyRequestedSchema);
+ long phase1Bytes = needBytes
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx,
keyOnlyRequestedSchema,
+ pushedFilterRanges)
+ : 0L;
+ PageReadStore keyPages = lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges);
+ if (keyPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block, and we already
+ // know pushedFilterRanges selects at least one row. Skipping the
block here would drop its
+ // surviving rows from the output, so assert rather than `continue`.
+ throw new IllegalStateException(
+ "No key pages for row group " + blockIdx + " despite "
+ + pushedFilterRanges.rowCount() + " rows selected by the
pushed filter");
+ }
+ RowRanges finalRanges = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+
+ if (finalRanges.rowCount() == 0) {
+ // Every surviving row was rejected by the storage filter; skip block
entirely. We still
+ // paid phase-1 to read the key column, so the bytes avoided vs a
no-storage-filter read
+ // are baseline - phase1 (the non-key bytes the no-filter path would
have read).
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcludedRg = m.rowsExcludedByRowGroup();
+ if (rowsExcludedRg != null) rowsExcludedRg.add(baselineRows);
+ if (bytesAvoidedRg != null) bytesAvoidedRg.add(baselineBytes -
phase1Bytes);
+ continue;
+ }
+
+ // Phase 2: switch to non-key schema, read non-key columns under
finalRanges. Skipped entirely
+ // when the projection is all keys (nonKeyRequestedSchema is null); emit
reconstructs each
+ // batch from the key queues alone.
+ long keptRows;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ if (nonKeyRequestedSchema == null) {
+ keptRows = finalRanges.rowCount();
+ phase2Bytes = 0L;
+ } else {
+ lateMatReader.setRequestedSchema(nonKeyRequestedSchema);
+ // requireOffsetIndexesForPhase2() already established that every
projected column of every
+ // row group has an offset index, so this page-filtering read cannot
fail for want of one.
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx, finalRanges);
+ if (dataPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block or empty ranges,
+ // both excluded above. Match phase 1 and fail with a message rather
than an NPE.
+ throw new IllegalStateException(
+ "No data pages for row group " + blockIdx + " despite " +
finalRanges.rowCount()
+ + " surviving rows");
+ }
+ keptRows = dataPages.getRowCount();
+ phase2Bytes = bytesAvoidedPf != null
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx,
nonKeyRequestedSchema,
+ finalRanges)
+ : 0L;
+ }
+ long filteredRows = baselineRows - keptRows;
+ SQLMetric rowsExcludedWithinRg = m.rowsExcludedWithinRowGroup();
+ if (rowsExcludedWithinRg != null && filteredRows > 0)
rowsExcludedWithinRg.add(filteredRows);
+ if (bytesAvoidedPf != null) {
+ bytesAvoidedPf.add(baselineBytes - phase1Bytes - phase2Bytes);
+ }
+
+ if (dataPages != null) {
+ if (rowIndexGenerator != null) {
+ rowIndexGenerator.initFromPageReadStore(dataPages);
+ }
+ for (int i = 0; i < columnVectors.length; i++) {
+ if (isKeyTopLevel[i]) {
+ // Key columns are sourced from the queues during emit; skip
phase-2 reader init.
+ continue;
+ }
+ initColumnReader(dataPages, columnVectors[i]);
+ }
+ }
+ totalCountLoadedSoFar += keptRows;
+ return;
+ }
+ hitEndOfData = true;
+ }
+
+ /**
+ * Compressed bytes the reader transfers for the leaf columns of {@code
schema} when it reads
+ * exactly {@code rowRanges} of the given block. Page headers and the
dictionary page are
+ * included, because both are read whenever any page of a chunk is read.
+ *
+ * <p>Two sources, chosen so this never causes IO of its own:
+ * <ul>
+ * <li>{@code rowRanges} covers the whole block: the answer is the sum of
the chunks'
+ * {@code getTotalSize()}, which is already in the footer. This is the
case that matters --
+ * whenever nothing else has built the block's {@link
ColumnIndexStore}, {@code rowRanges}
+ * is necessarily the whole block, because a narrower range can only
come from column-index
+ * filtering, which builds the store as a side effect.
+ * <li>{@code rowRanges} is a strict subset: walk the offset index, as
parquet's own read path
+ * does, and add the dictionary page the way {@code
calculateOffsetRanges} does. The store
+ * is guaranteed to exist here, so the walk is pure metadata
arithmetic.
+ * </ul>
+ *
+ * <p>Columns absent from this physical file (schema evolution) contribute
nothing, which is
+ * correct: the reader transfers nothing for them. Every caller for a given
block walks the same
+ * metadata, so a skipped column drops out of the baseline and the per-phase
totals alike.
+ */
+ private static long compressedBytesForRowRanges(
+ ParquetFileReader reader,
+ int blockIndex,
+ MessageType schema,
+ RowRanges rowRanges) {
+ if (schema == null || rowRanges.rowCount() == 0 ||
schema.getColumns().isEmpty()) {
+ return 0L;
+ }
+ BlockMetaData block = reader.getRowGroups().get(blockIndex);
+ long blockRowCount = block.getRowCount();
+ Map<ColumnPath, ColumnChunkMetaData> chunks = new HashMap<>();
+ for (ColumnChunkMetaData chunk : block.getColumns()) {
+ chunks.put(chunk.getPath(), chunk);
+ }
+ boolean wholeBlock = rowRanges.rowCount() == blockRowCount;
+ ColumnIndexStore ciStore = wholeBlock ? null :
reader.getColumnIndexStore(blockIndex);
+ long total = 0L;
+ for (ColumnDescriptor column : schema.getColumns()) {
+ ColumnPath path = ColumnPath.get(column.getPath());
+ ColumnChunkMetaData chunk = chunks.get(path);
+ if (chunk == null) {
+ // Column is in the (clipped) requested schema but not in this file.
+ continue;
+ }
+ if (wholeBlock) {
+ total += chunk.getTotalSize();
+ continue;
+ }
+ OffsetIndex offsetIndex;
+ try {
+ offsetIndex = ciStore.getOffsetIndex(path);
+ } catch (MissingOffsetIndexException e) {
+ continue;
+ }
+ if (offsetIndex == null) {
+ continue;
+ }
+ // The dictionary page is read whenever any data page of the chunk is,
so count it here the
+ // same way parquet's ColumnIndexFilterUtils.calculateOffsetRanges does.
+ total += dictionaryPageSize(chunk);
+ int pageCount = offsetIndex.getPageCount();
+ for (int i = 0; i < pageCount; i++) {
+ long from = offsetIndex.getFirstRowIndex(i);
+ long to = offsetIndex.getLastRowIndex(i, blockRowCount);
+ if (rowRanges.isOverlapping(from, to)) {
+ total += offsetIndex.getCompressedPageSize(i);
+ }
+ }
+ }
+ return total;
+ }
+
+ /**
+ * Compressed size of a chunk's dictionary page, or 0 if it has none.
+ * {@link ColumnChunkMetaData#getStartingPos()} already resolves to the
dictionary page offset
+ * when there is a valid one, so the gap up to the first data page is
exactly the dictionary page.
+ */
+ private static long dictionaryPageSize(ColumnChunkMetaData chunk) {
+ long startingPos = chunk.getStartingPos();
+ long firstDataPageOffset = chunk.getFirstDataPageOffset();
+ return startingPos < firstDataPageOffset ? firstDataPageOffset -
startingPos : 0L;
+ }
+
+ /**
+ * Reads all rows of the given key-only {@link PageReadStore} (which
contains only rows in
+ * {@code pushedFilterRanges}) in capacity-sized chunks, evaluates the
storage filter on each row,
+ * builds a {@link RowRanges} of surviving rows in original block-row
coordinates, and appends
+ * survivor key values into the per-key-column accumulators ({@link
#currentKeyAccumulators}).
+ * When an accumulator hits {@link #capacity}, it's pushed into {@link
#keyVectorQueues} and a
+ * fresh one is allocated. After all rows have been examined, any partial
trailing accumulator is
+ * pushed too.
+ *
+ * <p>The result is a subset of {@code pushedFilterRanges}: rows not in
{@code
+ * pushedFilterRanges} were never read and are implicitly excluded.
+ */
+ private RowRanges evaluateStorageFilter(
+ PageReadStore keyPages,
+ RowRanges pushedFilterRanges) throws IOException {
+ ensureKeyScratchAllocated();
+ VectorizedColumnReader[] readers = new
VectorizedColumnReader[keyDescriptors.length];
+ for (int i = 0; i < readers.length; i++) {
+ readers[i] = new VectorizedColumnReader(
+ keyDescriptors[i], keyRequired[i], keyPages, convertTz,
datetimeRebaseMode,
+ datetimeRebaseTz, int96RebaseMode, int96RebaseTz, writerVersion);
+ }
+ ensureCurrentKeyAccumulatorsAllocated();
+
+ long keyRowsTotal = pushedFilterRanges.rowCount();
+ PrimitiveIterator.OfLong rowIndexIter = pushedFilterRanges.iterator();
+ RowRanges.Builder finalRangesBuilder = RowRanges.builder();
+ long remaining = keyRowsTotal;
+ while (remaining > 0) {
+ int num = (int) Math.min((long) capacity, remaining);
+ for (int i = 0; i < keyScratchVectors.length; i++) {
+ keyScratchVectors[i].reset();
+ readers[i].readBatch(num, keyScratchVectors[i], null, null);
+ }
+ keyScratchBatch.setNumRows(num);
+ for (int r = 0; r < num; r++) {
+ long blockRow = rowIndexIter.nextLong();
+ if (storageFilter.test(keyScratchBatch.getRow(r))) {
+ finalRangesBuilder.addSelectedRow(blockRow);
+ appendSurvivorRowToAccumulators(r);
+ }
+ }
+ remaining -= num;
+ }
+
+ finalizePartialAccumulators();
+
+ return finalRangesBuilder.build();
+ }
+
+ private void ensureKeyScratchAllocated() {
+ if (keyScratchVectors != null) return;
+ keyScratchVectors = new WritableColumnVector[keyDescriptors.length];
+ boolean useOffHeap = MEMORY_MODE == MemoryMode.OFF_HEAP;
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ for (int i = 0; i < keyDescriptors.length; i++) {
+ DataType dt = sparkRequestedSchema.fields()[keyIndices[i]].dataType();
+ keyScratchVectors[i] = useOffHeap
+ ? new OffHeapColumnVector(capacity, dt)
+ : new OnHeapColumnVector(capacity, dt);
+ }
+ keyScratchBatch = new ColumnarBatch(keyScratchVectors);
+ }
+
+ /**
+ * Allocates the per-key-column accumulator vectors if any slot is null
(i.e. the previous
+ * accumulator was just pushed to the queue or this is the first row group).
Each accumulator has
+ * {@link #capacity} rows.
+ */
+ private void ensureCurrentKeyAccumulatorsAllocated() {
+ boolean useOffHeap = MEMORY_MODE == MemoryMode.OFF_HEAP;
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ for (int i = 0; i < currentKeyAccumulators.length; i++) {
+ if (currentKeyAccumulators[i] == null) {
+ DataType dt = sparkRequestedSchema.fields()[keyIndices[i]].dataType();
+ currentKeyAccumulators[i] = useOffHeap
+ ? new OffHeapColumnVector(capacity, dt)
+ : new OnHeapColumnVector(capacity, dt);
+ }
+ }
+ currentKeyAccumulatorRowCount = 0;
+ }
+
+ /**
+ * Appends row {@code srcRow} of each {@link #keyScratchVectors} into the
corresponding
+ * {@link #currentKeyAccumulators}. When the accumulators fill, they're
pushed onto their queues
+ * and fresh ones allocated. All key columns are appended in lockstep so
accumulators stay
+ * aligned.
+ */
+ private void appendSurvivorRowToAccumulators(int srcRow) {
+ final int dstRow = currentKeyAccumulatorRowCount;
+ final WritableColumnVector[] accs = currentKeyAccumulators;
+ final WritableColumnVector[] srcs = keyScratchVectors;
+ final ValueCopier[] copiers = keyCopiers;
+ for (int i = 0, n = accs.length; i < n; i++) {
+ WritableColumnVector src = srcs[i];
+ WritableColumnVector dst = accs[i];
+ if (src.isNullAt(srcRow)) {
+ dst.putNull(dstRow);
+ } else {
+ copiers[i].copy(dst, dstRow, src, srcRow);
+ }
+ }
+ currentKeyAccumulatorRowCount = dstRow + 1;
+ if (currentKeyAccumulatorRowCount == capacity) {
+ for (int i = 0; i < currentKeyAccumulators.length; i++) {
+ keyVectorQueues[i].addLast(currentKeyAccumulators[i]);
+ currentKeyAccumulators[i] = null;
+ }
+ ensureCurrentKeyAccumulatorsAllocated();
+ }
+ }
+
+ /**
+ * Pushes any partially-filled accumulator into its queue at row-group end
so the emit path can
+ * dequeue it as the row group's final batch.
+ */
+ private void finalizePartialAccumulators() {
+ if (currentKeyAccumulatorRowCount == 0) return;
+ for (int i = 0; i < currentKeyAccumulators.length; i++) {
+ keyVectorQueues[i].addLast(currentKeyAccumulators[i]);
+ currentKeyAccumulators[i] = null;
+ }
+ currentKeyAccumulatorRowCount = 0;
+ }
+
+ /**
+ * Closes anything held by the splicing path: pending dequeued key vectors
not yet rolled over,
+ * any vectors still queued (e.g. on early termination), and
partially-filled accumulators.
+ * Called from {@link #close()}.
+ */
+ private void closeSplicingState() {
+ if (pendingCloseKeyVectors != null) {
+ for (WritableColumnVector v : pendingCloseKeyVectors) {
+ if (v != null) v.close();
+ }
+ pendingCloseKeyVectors = null;
+ }
+ if (keyVectorQueues != null) {
+ for (java.util.ArrayDeque<WritableColumnVector> q : keyVectorQueues) {
+ if (q != null) {
+ for (WritableColumnVector v : q) v.close();
+ q.clear();
+ }
+ }
+ }
+ if (currentKeyAccumulators != null) {
+ for (WritableColumnVector v : currentKeyAccumulators) {
+ if (v != null) v.close();
+ }
+ currentKeyAccumulators = null;
+ }
+ }
+
+ /**
+ * Per-key-column value copier: appends one value from {@code src[srcRow]} to
+ * {@code dst[dstRow]}. Picked once at init via {@link
#copierFor(DataType)}; called per surviving
+ * row in {@link #appendSurvivorRowToAccumulators}. Caller handles null
sources.
+ */
+ @FunctionalInterface
+ private interface ValueCopier {
+ void copy(WritableColumnVector dst, int dstRow, WritableColumnVector src,
int srcRow);
+ }
+
+ /**
+ * Returns a {@link ValueCopier} for the given key {@link DataType}. The set
of types handled here
+ * is the definition behind {@code ParquetStorageFilter.isSupportedKeyType},
which gates both
+ * planning-time extraction and {@code ParquetStorageFilter.create} -- so
the throw at the end is
+ * unreachable. Teach both sides at once when adding a type; a type admitted
there but missing
+ * here becomes a task failure instead of a planning-time rejection.
+ */
+ private static ValueCopier copierFor(DataType dt) {
Review Comment:
The dead `VarcharType` and `CharType` branches are gone.
The dedup itself I would rather not do. `RowToColumnConverter` goes
`InternalRow` to vector, while these copiers go vector to vector, so reusing it
would put the row abstraction back in the innermost loop and, for a string, a
`UTF8String` wrapper per value. That is the opposite direction from your
`appendBytes` comment above, which asks the copy to get *more* specialized, and
that one is the follow-up I would rather take.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +929,450 @@ private void checkEndOfRowGroup() throws IOException {
totalCountLoadedSoFar += pages.getRowCount();
}
+ /**
+ * Loads the next row group using the three-phase late-materialization
pattern, all driven by the
+ * single {@link #lateMatReader} with its requested schema mutated per phase:
+ * - Phase 0 (full schema): compute {@code pushedFilterRanges} from the
pushed data filter via
+ * column index (metadata-only) using {@link
ParquetFileReader#getRowRanges}.
+ * - Phase 1 (key-only schema): read key-column pages restricted to {@code
pushedFilterRanges},
+ * evaluate the storage filter per row, build {@code finalRanges}.
+ * - Phase 2 (non-key schema): read non-key columns restricted to {@code
finalRanges}.
+ * Skipped entirely when {@link #nonKeyRequestedSchema} is null
(all-keys projection).
+ *
+ * Row groups for which {@code finalRanges} is empty are skipped entirely
(no phase-2 IO).
+ * Sets {@link #hitEndOfData} when all row groups have been processed.
+ */
+ private void loadNextRowGroupWithLateMaterialization() throws IOException {
+ while (nextBlockIndex < totalBlockCount) {
+ int blockIdx = nextBlockIndex++;
+ long blockRowCount =
lateMatReader.getRowGroups().get(blockIdx).getRowCount();
+ if (blockRowCount == 0) {
+ // parquet-mr never writes these, but RowRanges.createSingle(0) would
build Range(0, -1) and
+ // trip parquet's own `from <= to` assertion. The plain read path
skips them too.
+ continue;
+ }
+
+ // Phase 0: rows allowed by the pushed data filter (column-index
granularity). Restore the
+ // full requestedSchema first: phases 1 and 2 below narrow the reader's
schema, and
+ // ParquetFileReader.getRowRanges computes ranges against whatever
schema is set (it passes
+ // the reader's current `paths` to ColumnIndexFilter). This is defensive
-- with column-index
+ // filtering on, getFilteredRecordCount() in initialize() has already
memoized every block's
+ // ranges under the full schema, and with it off we do not call
getRowRanges at all.
+ lateMatReader.setRequestedSchema(requestedSchema);
+ // ParquetFileReader.getRowRanges only checks whether a filter is
pushed, NOT
+ // options.useColumnIndexFilter(), so calling it unconditionally would
keep applying
+ // column-index filtering after a user turned it off. That conf is the
documented escape hatch
+ // for files whose column index is wrong, and trusting a wrong column
index here would drop
+ // rows for good: finalRanges is a subset of pushedFilterRanges, and the
post-scan Filter no
+ // longer holds this predicate.
+ RowRanges pushedFilterRanges = useColumnIndexFilter
+ ? lateMatReader.getRowRanges(blockIdx)
+ : RowRanges.createSingle(blockRowCount);
+ if (pushedFilterRanges.rowCount() == 0) {
+ // Pushed data filter rejects this block entirely via column index.
Not a storage-filter
+ // skip, so we don't increment storage-filter metrics.
+ continue;
+ }
+
+ // Both byte metrics are always wired in production (FileSourceScanLike
creates all five
+ // whenever storageFilters is non-empty), so this only skips the work on
the test-only path
+ // that drives the reader directly. compressedBytesForRowRanges never
does IO of its own, so
+ // there is nothing here to avoid on the production path.
+ StorageFilterMetrics m = storageFilter.metrics();
+ SQLMetric bytesAvoidedRg = m.bytesAvoidedByRowGroup();
+ SQLMetric bytesAvoidedPf = m.bytesAvoidedByPageFiltering();
+ boolean needBytes = bytesAvoidedRg != null || bytesAvoidedPf != null;
+ long baselineRows = pushedFilterRanges.rowCount();
+ long baselineBytes = needBytes
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx,
requestedSchema,
+ pushedFilterRanges)
+ : 0L;
+
+ // Phase 1: switch to key-only schema, read key columns under
pushedFilterRanges, evaluate the
+ // storage filter per row.
+ lateMatReader.setRequestedSchema(keyOnlyRequestedSchema);
+ long phase1Bytes = needBytes
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx,
keyOnlyRequestedSchema,
+ pushedFilterRanges)
+ : 0L;
+ PageReadStore keyPages = lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges);
+ if (keyPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block, and we already
+ // know pushedFilterRanges selects at least one row. Skipping the
block here would drop its
+ // surviving rows from the output, so assert rather than `continue`.
+ throw new IllegalStateException(
+ "No key pages for row group " + blockIdx + " despite "
+ + pushedFilterRanges.rowCount() + " rows selected by the
pushed filter");
+ }
+ RowRanges finalRanges = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+
+ if (finalRanges.rowCount() == 0) {
+ // Every surviving row was rejected by the storage filter; skip block
entirely. We still
+ // paid phase-1 to read the key column, so the bytes avoided vs a
no-storage-filter read
+ // are baseline - phase1 (the non-key bytes the no-filter path would
have read).
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcludedRg = m.rowsExcludedByRowGroup();
+ if (rowsExcludedRg != null) rowsExcludedRg.add(baselineRows);
+ if (bytesAvoidedRg != null) bytesAvoidedRg.add(baselineBytes -
phase1Bytes);
+ continue;
+ }
+
+ // Phase 2: switch to non-key schema, read non-key columns under
finalRanges. Skipped entirely
+ // when the projection is all keys (nonKeyRequestedSchema is null); emit
reconstructs each
+ // batch from the key queues alone.
+ long keptRows;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ if (nonKeyRequestedSchema == null) {
+ keptRows = finalRanges.rowCount();
+ phase2Bytes = 0L;
+ } else {
+ lateMatReader.setRequestedSchema(nonKeyRequestedSchema);
+ // requireOffsetIndexesForPhase2() already established that every
projected column of every
+ // row group has an offset index, so this page-filtering read cannot
fail for want of one.
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx, finalRanges);
+ if (dataPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block or empty ranges,
+ // both excluded above. Match phase 1 and fail with a message rather
than an NPE.
+ throw new IllegalStateException(
+ "No data pages for row group " + blockIdx + " despite " +
finalRanges.rowCount()
+ + " surviving rows");
+ }
+ keptRows = dataPages.getRowCount();
+ phase2Bytes = bytesAvoidedPf != null
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx,
nonKeyRequestedSchema,
+ finalRanges)
+ : 0L;
+ }
+ long filteredRows = baselineRows - keptRows;
+ SQLMetric rowsExcludedWithinRg = m.rowsExcludedWithinRowGroup();
+ if (rowsExcludedWithinRg != null && filteredRows > 0)
rowsExcludedWithinRg.add(filteredRows);
+ if (bytesAvoidedPf != null) {
+ bytesAvoidedPf.add(baselineBytes - phase1Bytes - phase2Bytes);
+ }
+
+ if (dataPages != null) {
+ if (rowIndexGenerator != null) {
+ rowIndexGenerator.initFromPageReadStore(dataPages);
+ }
+ for (int i = 0; i < columnVectors.length; i++) {
+ if (isKeyTopLevel[i]) {
+ // Key columns are sourced from the queues during emit; skip
phase-2 reader init.
+ continue;
+ }
+ initColumnReader(dataPages, columnVectors[i]);
+ }
+ }
+ totalCountLoadedSoFar += keptRows;
+ return;
+ }
+ hitEndOfData = true;
+ }
+
+ /**
+ * Compressed bytes the reader transfers for the leaf columns of {@code
schema} when it reads
+ * exactly {@code rowRanges} of the given block. Page headers and the
dictionary page are
+ * included, because both are read whenever any page of a chunk is read.
+ *
+ * <p>Two sources, chosen so this never causes IO of its own:
+ * <ul>
+ * <li>{@code rowRanges} covers the whole block: the answer is the sum of
the chunks'
+ * {@code getTotalSize()}, which is already in the footer. This is the
case that matters --
+ * whenever nothing else has built the block's {@link
ColumnIndexStore}, {@code rowRanges}
+ * is necessarily the whole block, because a narrower range can only
come from column-index
+ * filtering, which builds the store as a side effect.
+ * <li>{@code rowRanges} is a strict subset: walk the offset index, as
parquet's own read path
+ * does, and add the dictionary page the way {@code
calculateOffsetRanges} does. The store
+ * is guaranteed to exist here, so the walk is pure metadata
arithmetic.
+ * </ul>
+ *
+ * <p>Columns absent from this physical file (schema evolution) contribute
nothing, which is
+ * correct: the reader transfers nothing for them. Every caller for a given
block walks the same
+ * metadata, so a skipped column drops out of the baseline and the per-phase
totals alike.
+ */
+ private static long compressedBytesForRowRanges(
Review Comment:
Fixed exactly as suggested: the three `List<ColumnDescriptor>` are fields
resolved once per file in `initializeLateMaterialization`, the reader is driven
through the `setRequestedSchema(List<ColumnDescriptor>)` overload, the
path-to-chunk map is built once per block and shared by the metric calls, and
both row counts are passed in instead of recomputed.
--
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]
