dongjoon-hyun commented on code in PR #58895:
URL: https://github.com/apache/spark/pull/58895#discussion_r4072847200
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -421,11 +552,313 @@ public boolean nextBatch() throws IOException {
return true;
}
+ /**
+ * Splicing emit path. Closes the previous emit's dequeued key vectors
(releasing survivor memory
+ * incrementally), advances to the next row group if needed via {@link
#checkEndOfRowGroup()},
+ * dequeues one survivor key vector per key column, drives non-key column
readers for {@code num}
+ * rows, and assembles a fresh {@link ColumnarBatch} interleaving key
(dequeued) and non-key
+ * (persistent value-vector) slots in the original projection order.
+ * The per-emit batch is a transient view over vectors owned elsewhere; see
{@link #close()} and
+ * {@link #closeSplicingState()}.
+ */
+ private boolean nextBatchSplicing() throws IOException {
+ if (pendingCloseKeyVectors != null) {
+ for (WritableColumnVector v : pendingCloseKeyVectors) {
+ if (v != null) v.close();
+ }
+ pendingCloseKeyVectors = null;
+ }
+ for (int i = 0; i < columnVectors.length; i++) {
+ if (isKeyTopLevel[i]) continue;
+ columnVectors[i].reset();
+ }
+ // Match the eager path and zero the outgoing batch before the terminal
checks below. Without
+ // this, a terminal call leaves `columnarBatch` pointing at the previous
emit whose key slots
+ // were just closed above -- with off-heap vectors those buffers are
already freed, so a
+ // consumer that read the batch after nextBatch() returned false would see
freed memory.
+ if (columnarBatch != null) columnarBatch.setNumRows(0);
+ if (hitEndOfData) return false;
+ if (rowsReturned >= totalRowCount) return false;
+ checkEndOfRowGroup();
+ if (hitEndOfData) return false;
+
+ int num = (int) Math.min(capacity, totalCountLoadedSoFar - rowsReturned);
+
+ WritableColumnVector[] dequeued = new
WritableColumnVector[keyVectorQueues.length];
+ for (int i = 0; i < keyVectorQueues.length; i++) {
+ dequeued[i] = keyVectorQueues[i].removeFirst();
Review Comment:
**Leak: these vectors are unreachable by `close()` until line 632.**
`removeFirst()` takes them out of `keyVectorQueues`, but
`pendingCloseKeyVectors = dequeued` is only reached at the end of the method.
If `columnReader.readBatch(...)` (L598) or `cv.assemble()` (L602) throws -- a
corrupt page, an object-store read error -- the vectors are in neither the
queues nor `pendingCloseKeyVectors`, so `close()` -> `closeSplicingState()`
walks only the remaining queue entries and skips them.
With `spark.sql.columnVector.offheap.enabled=true` these are
`Platform.allocateMemory` allocations freed only by `releaseMemory()` (no
finalizer, no `Cleaner`), so each of the 4 task attempts leaks `numKeys *
capacity` bytes permanently.
Moving `pendingCloseKeyVectors = dequeued;` to immediately after this loop
closes it.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -421,11 +552,313 @@ public boolean nextBatch() throws IOException {
return true;
}
+ /**
+ * Splicing emit path. Closes the previous emit's dequeued key vectors
(releasing survivor memory
+ * incrementally), advances to the next row group if needed via {@link
#checkEndOfRowGroup()},
+ * dequeues one survivor key vector per key column, drives non-key column
readers for {@code num}
+ * rows, and assembles a fresh {@link ColumnarBatch} interleaving key
(dequeued) and non-key
+ * (persistent value-vector) slots in the original projection order.
+ * The per-emit batch is a transient view over vectors owned elsewhere; see
{@link #close()} and
+ * {@link #closeSplicingState()}.
+ */
+ private boolean nextBatchSplicing() throws IOException {
+ if (pendingCloseKeyVectors != null) {
+ for (WritableColumnVector v : pendingCloseKeyVectors) {
+ if (v != null) v.close();
+ }
+ pendingCloseKeyVectors = null;
+ }
+ for (int i = 0; i < columnVectors.length; i++) {
+ if (isKeyTopLevel[i]) continue;
+ columnVectors[i].reset();
+ }
+ // Match the eager path and zero the outgoing batch before the terminal
checks below. Without
+ // this, a terminal call leaves `columnarBatch` pointing at the previous
emit whose key slots
+ // were just closed above -- with off-heap vectors those buffers are
already freed, so a
+ // consumer that read the batch after nextBatch() returned false would see
freed memory.
+ if (columnarBatch != null) columnarBatch.setNumRows(0);
+ if (hitEndOfData) return false;
+ if (rowsReturned >= totalRowCount) return false;
+ checkEndOfRowGroup();
+ if (hitEndOfData) return false;
+
+ int num = (int) Math.min(capacity, totalCountLoadedSoFar - rowsReturned);
+
+ WritableColumnVector[] dequeued = new
WritableColumnVector[keyVectorQueues.length];
+ for (int i = 0; i < keyVectorQueues.length; i++) {
+ dequeued[i] = keyVectorQueues[i].removeFirst();
+ }
+
+ for (int i = 0; i < columnVectors.length; i++) {
+ if (isKeyTopLevel[i]) continue;
+ ParquetColumnVector cv = columnVectors[i];
+ for (ParquetColumnVector leafCv : cv.getLeaves()) {
+ VectorizedColumnReader columnReader = leafCv.getColumnReader();
+ if (columnReader != null) {
+ columnReader.readBatch(num, leafCv.getValueVector(),
+ leafCv.getRepetitionLevelVector(),
leafCv.getDefinitionLevelVector());
+ }
+ }
+ cv.assemble();
+ }
+ if (rowIndexGenerator != null) {
+ // Row-index column is identified by name
(ROW_INDEX_TEMPORARY_COLUMN_NAME), which is a
+ // synthetic metadata column never referenced by a storage filter, so
its slot is a non-key
+ // slot with a persistent ParquetColumnVector.
+ rowIndexGenerator.populateRowIndex(columnVectors, num);
+ }
+
+ ColumnVector[] cols = new ColumnVector[persistentBatchColumns.length];
+ // This walks batch slots in ascending order while `keyIdx` walks the
survivor queues in
+ // key-row-position order, so it pairs the k-th smallest key slot with
key-row position k.
+ // That is only the identity because `ParquetStorageFilter.create` sorts
`keyColumnIndices`
+ // ascending -- see the comment there. `isKeyTopLevel` marks which slots
are keys but not their
+ // position in that list, so this loop cannot reconstruct the pairing on
its own: if the list
+ // ever stops being sorted, key columns silently swap places in the output
batch.
+ int keyIdx = 0;
+ for (int i = 0; i < persistentBatchColumns.length; i++) {
+ if (i < isKeyTopLevel.length && isKeyTopLevel[i]) {
+ cols[i] = dequeued[keyIdx++];
+ } else {
+ cols[i] = persistentBatchColumns[i];
+ }
+ }
+ columnarBatch = new ColumnarBatch(cols);
+ columnarBatch.setNumRows(num);
+
+ rowsReturned += num;
+ numBatched = num;
+ batchIdx = 0;
+ pendingCloseKeyVectors = dequeued;
+ return true;
+ }
+
private void initializeInternal() throws IOException,
UnsupportedOperationException {
missingColumns = new HashSet<>();
for (ParquetColumn column :
CollectionConverters.asJava(parquetColumn.children())) {
checkColumn(column);
}
+ if (storageFilter != null) {
+ initializeLateMaterialization();
+ }
+ }
+
+ /**
+ * Sets the storage filter for late materialization. Must be called before
{@link #initialize};
+ * {@link #initializeLateMaterialization()} (run from {@link #initialize})
inspects the per-file
+ * schema and decides whether splicing actually engages.
+ *
+ * <p>Splicing does NOT engage in one real case: all key columns are missing
from this physical
+ * file under schema evolution. The predicate is rewritten with each missing
key replaced by the
+ * constant the reader materializes for it (its existence DEFAULT, else null
-- see
+ * {@code ParquetStorageFilter.rewriteForMissingKeys}) and evaluated as a
constant: a true result
+ * keeps the file with no filtering, a false/null result skips it entirely.
Either way the rows
+ * the scan returns are exactly the rows that satisfy the filter, so this is
safe.
+ *
+ * <p>Every OTHER precondition is guaranteed by planning-time checks in
+ * {@code FileSourceStrategy.extractStorageFilters} and {@code
ParquetStorageFilter.create}, and
+ * violations throw rather than fall back: once extraction has moved a bloom
filter onto the scan
+ * it is gone from the post-scan Filter, so quietly not applying it would
produce wrong rows.
+ */
+ public void setStorageFilter(ParquetStorageFilter storageFilter) {
+ this.storageFilter = storageFilter;
+ }
+
+ private void initializeLateMaterialization() throws IOException {
+ lateMatReader = reader.getUnderlyingReader();
+ if (lateMatReader == null) {
+ // Unreachable in production: the only ParquetRowGroupReader
ParquetFileFormat builds is
+ // ParquetRowGroupReaderImpl, which returns its reader. Dropping the
filter here would return
+ // rows it rejects, since extraction already removed it from the
post-scan Filter.
+ throw new IllegalStateException(
+ "Storage-filter pushdown requires a reader backed by a
ParquetFileReader, but "
+ + reader.getClass().getName() + " does not expose one");
+ }
+ if (configuration != null) {
+ useColumnIndexFilter = configuration.getBoolean(
+ ParquetInputFormat.COLUMN_INDEX_FILTERING_ENABLED, true);
+ }
+
+ // Resolve each key column's top-level ParquetColumn. Partition into
present and missing (the
+ // latter can happen under schema evolution: a column is in the requested
schema but not in this
+ // physical parquet file). For any non-primitive key we still bail;
phase-1 reads only primitive
+ // leaves.
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ List<ParquetColumn> presentKeyColumns = new ArrayList<>(keyIndices.length);
+ List<Integer> missingKeyLocalPositions = new ArrayList<>();
+ for (int i = 0; i < keyIndices.length; i++) {
+ int idx = keyIndices[i];
+ if (idx < 0 || idx >= parquetColumn.children().size()) {
+ // Unreachable: ParquetStorageFilter.create rejects out-of-range
ordinals. Fail loudly
+ // rather than dropping the filter -- extractStorageFilters already
removed it from the
+ // post-scan Filter, so silently ignoring it here would return wrong
rows.
+ throw new IllegalStateException(String.format(
+ "Storage-filter key ordinal %d is out of range for a %d-column
requested schema",
+ idx, parquetColumn.children().size()));
+ }
+ ParquetColumn column = parquetColumn.children().apply(idx);
+ if (!column.isPrimitive()) {
+ // Unreachable: ParquetStorageFilter.isSupportedKeyType admits only
types with a primitive
+ // Parquet leaf, and it gates both planning and
ParquetStorageFilter.create. Fail loudly for
+ // the same reason as above.
+ throw new IllegalStateException(
+ "Storage-filter key column is not a primitive Parquet column: " +
column.path());
+ }
+ if (missingColumns.contains(column)) {
+ missingKeyLocalPositions.add(i);
+ } else {
+ presentKeyColumns.add(column);
+ }
+ }
+
+ // If any key column is missing from this file, rewrite the predicate to
substitute the constant
+ // the reader will actually materialize for that column. That is the
column's existence DEFAULT
+ // when it has one (ParquetColumnVector writes it into the output vector
and marks the vector
+ // constant), otherwise null. Substituting null for a column that reads
back as its default
+ // would filter on a value the scan never returns. The predicate must be
evaluated against the
+ // substituted constant rather than skipped: null does not always mean
false in a filter.
+ if (!missingKeyLocalPositions.isEmpty()) {
+ int[] missing = new int[missingKeyLocalPositions.size()];
+ Object[] missingValues = new Object[missingKeyLocalPositions.size()];
+ Object[] existenceDefaults =
+ ResolveDefaultColumns.existenceDefaultValues(sparkRequestedSchema);
+ for (int i = 0; i < missing.length; i++) {
+ missing[i] = missingKeyLocalPositions.get(i);
+ missingValues[i] = existenceDefaults[keyIndices[missing[i]]];
+ }
+ storageFilter = storageFilter.rewriteForMissingKeys(missing,
missingValues);
+
+ if (presentKeyColumns.isEmpty()) {
+ // All key columns missing: the rewritten predicate is fully constant.
Evaluate once and
+ // apply uniformly to the whole file.
+ boolean keepAll = storageFilter.evalAllMissing();
+ if (keepAll) {
+ // Predicate is constant-true for this file: no filtering to do.
+ storageFilter = null;
+ } else {
+ // Predicate is constant-false/null: no row from this file can pass
the filter.
+ storageFilter = null;
+ hitEndOfData = true;
+ }
+ return;
+ }
+ }
+
+ keyDescriptors = new ColumnDescriptor[presentKeyColumns.size()];
+ keyRequired = new boolean[presentKeyColumns.size()];
+ Types.MessageTypeBuilder keySchemaBuilder = Types.buildMessage();
+ Set<String> keyTopLevelNames = new HashSet<>();
+ for (int i = 0; i < presentKeyColumns.size(); i++) {
+ ParquetColumn column = presentKeyColumns.get(i);
+ keyDescriptors[i] = column.descriptor().get();
+ keyRequired[i] = column.required();
+ // Preserve the field name/type as it appears at the top of
requestedSchema.
+ String topLevelName = keyDescriptors[i].getPath()[0];
+ keySchemaBuilder.addField(requestedSchema.getType(topLevelName));
+ keyTopLevelNames.add(topLevelName);
+ }
+ keyOnlyRequestedSchema = keySchemaBuilder.named(requestedSchema.getName());
+
+ // Build the non-key (complement) schema. When the projection has at least
one non-key column,
+ // phase 2 will switch lateMatReader's schema to this and read those
columns under finalRanges.
+ // When the projection is *all* key columns (e.g. a scan whose only column
is the bloom probe),
+ // splicing still pays off: phase 2 has nothing useful to read, so we skip
it entirely (no read,
+ // no IO) and emit batches purely from the key queues. The
`nonKeyRequestedSchema != null` check
+ // downstream gates phase-2 IO.
+ Types.MessageTypeBuilder nonKeyBuilder = Types.buildMessage();
+ int nonKeyFieldCount = 0;
+ for (Type field : requestedSchema.getFields()) {
+ if (!keyTopLevelNames.contains(field.getName())) {
+ nonKeyBuilder.addField(field);
+ nonKeyFieldCount++;
+ }
+ }
+ if (nonKeyFieldCount > 0) {
+ nonKeyRequestedSchema = nonKeyBuilder.named(requestedSchema.getName());
+ requireOffsetIndexesForPhase2();
+ }
+ initializeSplicingState(presentKeyColumns);
+
+ totalBlockCount = lateMatReader.getRowGroups().size();
+ nextBlockIndex = 0;
+ }
+
+ /**
+ * Fails now if any projected column of any row group lacks a Parquet offset
index.
+ *
+ * <p>Phase 2 reads a strict subset of a row group's rows, which parquet can
only do via the
+ * offset index; files written before parquet-mr 1.11, or by writers that
omit it, have none. We
+ * cannot
+ * widen phase 2 to the whole block instead, because the key vectors already
hold only the
+ * survivors and the batch would misalign -- and we cannot skip the filter
either, since
+ * {@code extractStorageFilters} has already removed it from the post-scan
Filter.
+ *
+ * <p>Every projected column is checked, not just the non-key ones phase 2
reads, because parquet
+ * builds one column index store per row group and reuses it. Phase 0 asks
for the row ranges
+ * under the full requested schema, so {@code ColumnIndexStoreImpl.create}
is called with the key
+ * columns in its path set, and it returns its {@code EMPTY} singleton as
soon as any one of those
+ * paths has no offset index. {@code ParquetFileReader.getColumnIndexStore}
memoizes that store
+ * per block, and {@code EMPTY.getOffsetIndex} throws for *every* column. So
a key column with no
+ * offset index kills phase 2 too, with a raw {@code
MissingOffsetIndexException} naming some
+ * non-key column and none of the guidance below.
+ *
+ * <p>Checking up front rather than at the first partially-kept row group is
deliberate: whether
+ * phase 2 needs the offset index otherwise depends on how selective the
filter turns out to be on
+ * this particular file, so the same query would fail or not depending on
the data. This is
+ * conservative -- a filter that happens to keep every row of every block
would not have needed
+ * the offset index -- but such a filter also saves nothing, so failing
loudly loses nothing.
+ *
+ * <p>The check itself is free: {@code getOffsetIndexReference()} is a
footer field that
+ * {@link #initialize} has already read.
+ */
+ private void requireOffsetIndexesForPhase2() {
+ Set<ColumnPath> projectedPaths = new HashSet<>();
+ for (ColumnDescriptor column : requestedSchema.getColumns()) {
+ projectedPaths.add(ColumnPath.get(column.getPath()));
+ }
+ List<BlockMetaData> blocks = lateMatReader.getRowGroups();
+ for (int blockIdx = 0; blockIdx < blocks.size(); blockIdx++) {
+ for (ColumnChunkMetaData chunk : blocks.get(blockIdx).getColumns()) {
+ if (projectedPaths.contains(chunk.getPath()) &&
chunk.getOffsetIndexReference() == null) {
Review Comment:
**This turns a performance feature into an availability one.**
Any Parquet file without an offset index fails the query outright, with no
per-file fallback. That is not an exotic shape: pyarrow's `write_table`
defaults to `write_page_index=False`, Impala omits it, and parquet-mr < 1.11
(Spark <= 2.4) has no column/offset index at all.
Scenario: an operator sets
`spark.sql.parquet.storageFilterPushdown.enabled=true` as a cluster default.
Any join where `InjectRuntimeFilter` attaches a bloom to such a table now
throws from `initialize()` on every task, retries 4x, and fails the stage --
where before this PR the bloom was an ordinary post-scan `FilterExec` and the
query worked. One legacy file in one partition is enough.
Two things make it sting more than it needs to: the check is eager over
every row group even when the filter keeps 100% of rows and phase 2 would never
consult the index, and the only remedy the message offers is a global conf flip.
The reader already has precedent for degrading per file -- L731-741 sets
`storageFilter = null` when every key column is missing. The same shape would
work here: fall back to the eager read for this file and evaluate the predicate
during splice. That keeps the "extraction already removed the conjunct"
invariant intact while leaving the feature purely optional.
##########
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() {
Review Comment:
**If this allocation loop throws, `close()` cannot reach what it already
allocated.**
`keyScratchBatch` is assigned only after the loop completes, and `close()`
(L338-342) frees the scratch vectors solely through `keyScratchBatch`. So an
off-heap allocation failure on the second iteration leaves `keyScratchVectors`
non-null holding a live `OffHeapColumnVector` while `keyScratchBatch` is still
null -- `close()` tests `keyScratchBatch != null`, finds null, and skips them.
Secondarily, the `if (keyScratchVectors != null) return;` guard means a
later call would proceed with a partially populated array and NPE at
`readers[i].readBatch(num, keyScratchVectors[i], ...)`.
Closing `keyScratchVectors` element-wise, the way `closeSplicingState()`
already does for the accumulators, removes both.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -421,11 +552,313 @@ public boolean nextBatch() throws IOException {
return true;
}
+ /**
+ * Splicing emit path. Closes the previous emit's dequeued key vectors
(releasing survivor memory
+ * incrementally), advances to the next row group if needed via {@link
#checkEndOfRowGroup()},
+ * dequeues one survivor key vector per key column, drives non-key column
readers for {@code num}
+ * rows, and assembles a fresh {@link ColumnarBatch} interleaving key
(dequeued) and non-key
+ * (persistent value-vector) slots in the original projection order.
+ * The per-emit batch is a transient view over vectors owned elsewhere; see
{@link #close()} and
+ * {@link #closeSplicingState()}.
+ */
+ private boolean nextBatchSplicing() throws IOException {
+ if (pendingCloseKeyVectors != null) {
+ for (WritableColumnVector v : pendingCloseKeyVectors) {
+ if (v != null) v.close();
+ }
+ pendingCloseKeyVectors = null;
+ }
+ for (int i = 0; i < columnVectors.length; i++) {
+ if (isKeyTopLevel[i]) continue;
+ columnVectors[i].reset();
+ }
+ // Match the eager path and zero the outgoing batch before the terminal
checks below. Without
+ // this, a terminal call leaves `columnarBatch` pointing at the previous
emit whose key slots
+ // were just closed above -- with off-heap vectors those buffers are
already freed, so a
+ // consumer that read the batch after nextBatch() returned false would see
freed memory.
+ if (columnarBatch != null) columnarBatch.setNumRows(0);
+ if (hitEndOfData) return false;
+ if (rowsReturned >= totalRowCount) return false;
+ checkEndOfRowGroup();
+ if (hitEndOfData) return false;
+
+ int num = (int) Math.min(capacity, totalCountLoadedSoFar - rowsReturned);
+
+ WritableColumnVector[] dequeued = new
WritableColumnVector[keyVectorQueues.length];
+ for (int i = 0; i < keyVectorQueues.length; i++) {
+ dequeued[i] = keyVectorQueues[i].removeFirst();
+ }
+
+ for (int i = 0; i < columnVectors.length; i++) {
+ if (isKeyTopLevel[i]) continue;
+ ParquetColumnVector cv = columnVectors[i];
+ for (ParquetColumnVector leafCv : cv.getLeaves()) {
+ VectorizedColumnReader columnReader = leafCv.getColumnReader();
+ if (columnReader != null) {
+ columnReader.readBatch(num, leafCv.getValueVector(),
+ leafCv.getRepetitionLevelVector(),
leafCv.getDefinitionLevelVector());
+ }
+ }
+ cv.assemble();
+ }
+ if (rowIndexGenerator != null) {
+ // Row-index column is identified by name
(ROW_INDEX_TEMPORARY_COLUMN_NAME), which is a
+ // synthetic metadata column never referenced by a storage filter, so
its slot is a non-key
+ // slot with a persistent ParquetColumnVector.
+ rowIndexGenerator.populateRowIndex(columnVectors, num);
+ }
+
+ ColumnVector[] cols = new ColumnVector[persistentBatchColumns.length];
+ // This walks batch slots in ascending order while `keyIdx` walks the
survivor queues in
+ // key-row-position order, so it pairs the k-th smallest key slot with
key-row position k.
+ // That is only the identity because `ParquetStorageFilter.create` sorts
`keyColumnIndices`
+ // ascending -- see the comment there. `isKeyTopLevel` marks which slots
are keys but not their
+ // position in that list, so this loop cannot reconstruct the pairing on
its own: if the list
+ // ever stops being sorted, key columns silently swap places in the output
batch.
+ int keyIdx = 0;
+ for (int i = 0; i < persistentBatchColumns.length; i++) {
+ if (i < isKeyTopLevel.length && isKeyTopLevel[i]) {
+ cols[i] = dequeued[keyIdx++];
+ } else {
+ cols[i] = persistentBatchColumns[i];
+ }
+ }
+ columnarBatch = new ColumnarBatch(cols);
Review Comment:
**This breaks `resultBatch()`'s documented contract.**
`resultBatch()`'s javadoc says "Returns the ColumnarBatch object that will
be used for all rows returned by this reader. **This object is reused.**" Under
splicing a new `ColumnarBatch` is created per batch, so a caller that follows
the documented hoist-once pattern reads zero rows forever -- no exception, no
wrong values, just silently empty.
That pattern exists in-tree: `DataSourceReadBenchmark.scala:234` and
`ParquetEncodingSuite.scala:74` both do `val batch = reader.resultBatch()`
outside the `while (reader.nextBatch())` loop. Neither uses a storage filter
today, and production v1 paths are safe because `RecordReaderIterator.next()`
re-reads `getCurrentValue` per batch -- but `resultBatch()` is public and this
suite's own helpers deliberately re-fetch it inside the loop, which suggests
the constraint was already noticed.
Either update the javadoc, or keep one stable `ColumnarBatch` and mutate its
column array in place.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -151,6 +152,63 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
}
}
+ /**
+ * Splits `afterScanFilters` into bloom-filter conjuncts that can be pushed
to the storage layer
+ * for late materialization (returned as the first element) and the
remaining filters that stay as
+ * a post-scan FilterExec (returned as the second element).
+ *
+ * Eligibility (statically checked here so the runtime never silently loses
the filter):
+ * - The storage-filter pushdown SQL conf is on.
+ * - The file format is exactly [[ParquetFileFormat]]. Subclasses are
excluded on purpose: they
+ * may customize reading by overriding `buildReaderWithPartitionValues`,
and attaching storage
+ * filters would route the scan through `ParquetFileFormat`'s own reader
instead, silently
+ * dropping whatever the subclass does.
+ * - The vectorized reader is feasible for the schema the reader will
actually see, i.e.
+ * `partitionSchema ++ outputDataSchema` -- the same schema
`ParquetFileFormat.buildReader`
+ * derives `enableVectorizedReader` from.
+ * - The conjunct is a top-level [[BloomFilterMightContain]] (not nested
under OR/NOT).
+ * - The conjunct is deterministic. `ParquetStorageFilter.test` evaluates
the predicate without
+ * calling `BasePredicate.initialize(partitionIndex)`, which
`GeneratePredicate` emits for a
+ * `Nondeterministic` expression, so a non-deterministic conjunct would
fail at task time. No
+ * such bloom exists today, because the only producer is
`InjectRuntimeFilter` and a join key
+ * is deterministic, but this gate should not depend on a distant rule.
+ * - The bloom's value-side references are projected data columns whose
type the reader's value
+ * copier supports (see [[ParquetStorageFilter.isSupportedKeyType]]).
+ *
+ * If any condition fails, ALL bloom filters stay in the second element to
preserve the existing
+ * fallback behavior.
+ */
+ private def extractStorageFilters(
+ afterScanFilters: ExpressionSet,
+ fsRelation: HadoopFsRelation,
+ readDataColumns: Seq[Attribute],
+ outputDataSchema: StructType): (Seq[Expression], ExpressionSet) = {
+ val sparkSession = fsRelation.sparkSession
+ val sqlConf = sparkSession.sessionState.conf
+ if (!sqlConf.parquetStorageFilterPushdownEnabled) return (Nil,
afterScanFilters)
+ if (fsRelation.fileFormat.getClass != classOf[ParquetFileFormat]) {
Review Comment:
**Could the format answer this instead of the planner type-checking it?**
This makes the generic `FileSourceStrategy` import `ParquetFileFormat` and
`ParquetStorageFilter` (L36) and hard-code an exact-class check, so ORC or any
third-party format can only participate by editing the planner, and
`ParquetFileFormat` subclasses have no opt-in at all.
L205 goes a step further: `ParquetStorageFilter.isSupportedKeyType` is the
set of types `VectorizedParquetRecordReader.copierFor` can copy -- a reader
implementation detail the planner now owns, with a "must stay in lockstep"
invariant spanning two packages and enforced only by comment.
Spark's established shape for this is a capability method on `FileFormat`,
exactly like `supportBatch`, `supportDataType`, `vectorTypes` and
`metadataSchemaFields` -- and `supportBatch` is already being called
polymorphically five lines below. Something like:
```scala
def supportsStorageFilter(expr: Expression, readDataSchema: StructType):
Boolean = false
```
would absorb both checks and keep the type list inside the parquet package.
The justification given above is that subclasses may override
`buildReaderWithPartitionValues` -- but that concern is a consequence of this
PR inverting the delegation, and it can be expressed inside the override, in
the class that knows why: `getClass == classOf[ParquetFileFormat] && ...`.
##########
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:
**`MessageType.getColumns()` is rebuilt ~9x per row group, for two metrics.**
`getColumns()` is not cached in parquet-mr (its source carries a `// TODO:
optimize this`): each call rebuilds `getPaths(0)` and, per leaf, does
`getType(path)` + `getMaxRepetitionLevel` + `getMaxDefinitionLevel` tree
descents plus a `new ColumnDescriptor`.
`ParquetFileReader.setRequestedSchema(MessageType)` is
`setRequestedSchema(projection.getColumns())`, so L961/L993/L1031 contribute
three rebuilds per row group. This method calls `schema.getColumns()` twice per
invocation (L1099, L1111) across three invocations (L986, L995, L1043) for six
more. It also rebuilds a `HashMap<ColumnPath, ColumnChunkMetaData>` over
*every* chunk of the block rather than the projected ones (L1104-1107), and
`rowRanges.rowCount()` -- O(#ranges), and `finalRanges` can hold ~250K ranges
for a 1M-row block at 50% selectivity -- is recomputed at L1009, L1099 and
L1108.
For a 200-column projection on a 2000-column file that is roughly 2,400
allocations, 4,800 tree descents and 6,000 `HashMap.put`s per row group, to
populate two counters.
Hoisting the three `List<ColumnDescriptor>` into fields in
`initializeLateMaterialization()` (and using the
`setRequestedSchema(List<ColumnDescriptor>)` overload), building the
path->chunk map once per block, and passing the already-computed row counts in
removes all of it with no behavior change.
##########
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:
**This duplicates `RowToColumnConverter`, and has already drifted.**
`RowToColumnConverter.getConverterForType`
(`sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala:288-327`)
has the identical groupings -- `IntegerType | DateType | _:
YearMonthIntervalType` -> Int, `LongType | TimestampType | TimestampNTZType |
_: DayTimeIntervalType | _: TimeType` -> Long, and the same `MAX_INT_DIGITS` /
`MAX_LONG_DIGITS` three-way decimal split. `BasicNullableTypeConverter` also
already handles the `isNullAt` -> `putNull` branch at L1249.
The hot loop already has an `InternalRow` in hand
(`keyScratchBatch.getRow(r)`), so `convert(row, currentKeyAccumulators)` is
close to a drop-in, and `ColumnVectorUtils.java:134-138` is the existing
precedent for calling this `private[execution]` Scala class from Java in this
module.
The drift is visible in the next block: the `VarcharType` / `CharType`
branches at L1367-1368 are dead, since both extend `StringType`
(`sql/api/.../CharType.scala:34`, `VarcharType.scala:33`) and the `StringType`
test comes first in the same `||` chain.
`ParquetStorageFilter.isSupportedKeyType`'s comment at L216 states this
correctly, so the two files already disagree about the same fact -- which is
the "keep them in lockstep" cost the javadoc above warns about.
##########
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:
**These two fields are never read, and the javadoc is inaccurate.**
Repo-wide: `inputFile` is assigned at L120, L125 and L193 and read nowhere;
`fileFooter`'s only read is `fileFooter.getFileMetaData().getSchema()` at L197,
inside the same method that assigns it, where the pre-existing
`fileReader.getFooter()` expression was equivalent. The late-materialization
path gets what it needs from `reader.getUnderlyingReader()` /
`lateMatReader.getFile()` instead.
The doc also says "Set by both `initialize` overloads" -- there are four,
and the `@VisibleForTesting initialize(MessageType, MessageType,
ParquetRowGroupReader, int)` at L226 (the one
`ParquetVectorizedSuite.scala:802` uses) leaves both null, so a subclass
trusting the doc would NPE.
One further side effect: `close()` nulls only `reader`, so `fileFooter` now
pins the whole `ParquetMetadata` -- every `BlockMetaData` and column-chunk
statistic -- for the reader's lifetime on *every* Parquet scan, not just
storage-filter ones.
Dropping the hunk looks right; `fileFooter` can stay a local.
##########
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();
Review Comment:
**Unbounded, and invisible to Spark's memory manager.**
This loop evaluates the entire row group before
`loadNextRowGroupWithLateMaterialization` returns, so `keyVectorQueues` holds
every surviving key value of one row group at once -- and those vectors are
live at the same time as phase 2's non-key vectors.
At the default 128MB `parquet.block.size` a row group is commonly 1M+ rows,
and blooms are deliberately false-positive-prone, so most survive. For a
32-byte string key at `capacity = 4096` that is ~250 full-width vectors plus
their variable-length child buffers, per key column, per concurrent task. With
off-heap enabled these go through `Platform.allocateMemory` rather than a
`MemoryConsumer`, so the executor gets OOM-killed by YARN/k8s instead of
raising `SparkOutOfMemoryError` or spilling.
The conf doc's "a task holds up to one extra copy of the key columns for one
row group" is accurate but reads as bounded. It scales with row-group size, not
with `capacity`, and for a dictionary-encoded key the accumulators store
*expanded* bytes -- a chunk holding 1,000 distinct 40-byte values plus 1M
dictionary IDs becomes ~20MB of materialized strings.
Evaluating one `capacity`-sized chunk at a time (emit, then continue) would
bound this and fix the `LIMIT` stall below; a guard that falls back to the
eager path above some `blockRowCount` would be the cheaper stopgap.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -0,0 +1,1965 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.io.{ByteArrayOutputStream, File}
+import java.net.URI
+import java.time.LocalTime
+import java.util.concurrent.atomic.AtomicLong
+
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileStatus, FSDataInputStream, FSInputStream,
Path, RawLocalFileSystem}
+import org.apache.hadoop.mapreduce.Job
+import org.apache.parquet.hadoop.{ParquetInputFormat, ParquetOutputFormat}
+
+import org.apache.spark.paths.SparkPath
+import org.apache.spark.sql.{sources, QueryTest, Row, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute,
BloomFilterMightContain, BoundReference, Coalesce, Expression,
GreaterThanOrEqual, IsNull, LessThanOrEqual, Literal, Or, Predicate, Rand,
XxHash64}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter => LogicalFilter}
+import org.apache.spark.sql.execution.{CollapseCodegenStages,
ColumnarToRowExec, FileSourceScanExec, FileSourceScanLike, FilterExec,
LocalLimitExec, SparkPlan, WholeStageCodegenExec}
+import org.apache.spark.sql.execution.datasources.{FileFormat,
FileSourceStrategy, OutputWriterFactory, PartitionedFile}
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
+import org.apache.spark.util.sketch.BloomFilter
+
+/**
+ * Tests the late-materialization path of [[VectorizedParquetRecordReader]]
driven by a
+ * [[ParquetStorageFilter]]. Writes small multi-row-group parquet files, wires
a hand-built filter
+ * into the reader, and asserts correctness + the two storage-filter metrics.
+ */
+class ParquetStorageFilterSuite extends QueryTest with SharedSparkSession {
+ import testImplicits._
+
+ // Writes a parquet file with the given rows and row-group size; returns the
path.
+ private def writeParquetFile(
+ dir: File,
+ rows: Seq[(Long, String)],
+ rowGroupSize: Long = 1024L,
+ pageSize: Option[Long] = None): String = {
+ val outDir = new File(dir, s"test-${System.nanoTime()}").getAbsolutePath
+ val writer = rows.toDF("k", "v")
+ .repartition(1)
+ .write
+ .option(ParquetOutputFormat.BLOCK_SIZE, rowGroupSize)
+ // Dictionary encoding off keeps row-group sizing predictable. Note this
does NOT disable the
+ // column index, despite what an earlier version of this comment claimed.
+ .option(ParquetOutputFormat.ENABLE_DICTIONARY, "false")
+ // A small page size gives each row group several pages per column, which
is what lets
+ // column-index filtering produce a row range narrower than the whole row
group.
+ pageSize.foreach(size => writer.option(ParquetOutputFormat.PAGE_SIZE,
size))
+ writer.parquet(outDir)
+ val files = new File(outDir).listFiles((_, name) =>
name.endsWith(".parquet"))
+ assert(files != null && files.length == 1, s"expected exactly one parquet
file under $outDir")
+ files(0).getAbsolutePath
+ }
+
+ // Collects all rows from a reader initialized with the given storage filter.
+ private def readAll(
+ filePath: String,
+ storageFilter: ParquetStorageFilter): (Seq[(Long, String)],
VectorizedParquetRecordReader) = {
+ val reader = new VectorizedParquetRecordReader(false, 4096)
+ reader.setStorageFilter(storageFilter)
+ reader.initialize(filePath, java.util.Arrays.asList("k", "v"))
+ reader.initBatch(new StructType(), null)
+ val collected = mutable.ArrayBuffer[(Long, String)]()
+ while (reader.nextBatch()) {
+ val batch = reader.resultBatch().asInstanceOf[ColumnarBatch]
+ val n = batch.numRows()
+ val kVec = batch.column(0)
+ val vVec = batch.column(1)
+ var i = 0
+ while (i < n) {
+ collected += ((kVec.getLong(i), vVec.getUTF8String(i).toString))
+ i += 1
+ }
+ }
+ (collected.toSeq, reader)
Review Comment:
**The drain happens outside the callers' `try`/`finally`, so the reader
leaks on exactly the failures these tests exist to catch.**
This helper (and `readKeyOnlyAll` L152, `readAllWith` L771,
`survivorsViaPlainPath` L805) runs the whole `while (reader.nextBatch())` loop
and only then returns the reader. Callers wrap just the assertions in `try {
... } finally { reader.close() }`, so the `finally` protects nothing that can
realistically fail.
Concretely: the off-heap parameterized test at L1021-1041 calls
`readAllWith(..., useOffHeap = true)`. A misaligned-queue regression would
surface as `NoSuchElementException` from `keyVectorQueues[i].removeFirst()`
inside `nextBatch()` -- and then the reader is never closed, leaking the
`ParquetFileReader`, its `SeekableInputStream`, the off-heap
`persistentBatchColumns`, the survivor queues and the scratch vectors for the
rest of the JVM, which can cascade into unrelated failures in the same suite.
`survivorsViaPlainPath` is the same shape and is the oracle for ~35
parameterized key-type tests, and its `extract` can `fail(...)` from inside the
loop.
Wrapping the drain in `Utils.tryWithSafeFinally`, or having the helper own
`close()` and return only the rows, fixes all four.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -165,6 +166,47 @@ trait FileFormat {
}
}
+ /**
+ * Like [[buildReaderWithPartitionValues]] but additionally accepts a
sequence of storage filters:
+ * Catalyst expressions that the storage layer may evaluate to drive
value-column IO pruning based
+ * on key-column evaluation (e.g., late materialization with a runtime bloom
filter).
+ *
+ * The default implementation delegates to
[[buildReaderWithPartitionValues]] and accepts no
+ * storage filter at all. File formats that support storage-filter pushdown
(e.g., Parquet) should
+ * override this method.
+ *
+ * A non-empty `storageFilters` here is a planner bug, so the default body
rejects it rather than
+ * dropping it. The planner removes an extracted conjunct from the post-scan
`Filter`, so a reader
+ * that ignores it returns rows the filter rejects. Every other layer of the
feature fails loudly
+ * for the same reason. Unreachable today, because
`FileSourceStrategy.extractStorageFilters` only
+ * extracts for `ParquetFileFormat` itself, but that keeps the invariant one
method away from the
+ * code that relies on it.
+ *
+ * Scalar subqueries inside `storageFilters` are expected to have been
materialized before this
+ * method is called, so that the returned reader can be safely serialized to
executors.
+ *
+ * `storageFilterMetrics` is an optional map of SQL metrics the reader can
update during execution
+ * (e.g. number of row groups skipped). The scan is expected to expose these
metrics via its
+ * `metrics` field so they show up in the SQL UI.
+ */
+ def buildReaderWithStorageFilters(
+ sparkSession: SparkSession,
+ dataSchema: StructType,
+ partitionSchema: StructType,
+ requiredSchema: StructType,
+ filters: Seq[Filter],
+ storageFilters: Seq[Expression],
+ options: Map[String, String],
+ hadoopConf: Configuration,
+ storageFilterMetrics: Map[String, SQLMetric] = Map.empty
+ ): PartitionedFile => Iterator[InternalRow] = {
+ require(storageFilters.isEmpty,
+ s"${getClass.getSimpleName} does not support storage-filter pushdown,
but was given " +
+ storageFilters.mkString("[", ", ", "]"))
+ buildReaderWithPartitionValues(
Review Comment:
**Mutual recursion with `ParquetFileFormat`.**
This default calls `buildReaderWithPartitionValues`, and
`ParquetFileFormat.buildReaderWithPartitionValues` (L195) now delegates to
`buildReaderWithStorageFilters`. That call is unqualified and neither method is
`final`, so it dispatches virtually.
A subclass of `ParquetFileFormat` that opts out of storage filters with the
most natural body -- the same one this default uses:
```scala
override def buildReaderWithStorageFilters(...) =
buildReaderWithPartitionValues(sparkSession, dataSchema, partitionSchema,
requiredSchema, filters, options, hadoopConf)
```
gets `buildReaderWithPartitionValues` -> subclass override ->
`buildReaderWithPartitionValues` -> ... and a `StackOverflowError` on the
driver the first time `FileSourceScanExec.inputRDD` builds the reader. Even
`super.buildReaderWithPartitionValues(...)` recurses, because the super body's
unqualified call still lands on the subclass override -- only
`super.buildReaderWithStorageFilters(...)` is safe, and the scaladoc does not
say so while actively inviting the override ("File formats that support
storage-filter pushdown ... should override this method").
Not reachable in-tree, but subclassing `ParquetFileFormat` is a common
downstream pattern. Routing `ParquetFileFormat`'s base implementation through a
private method, or marking one of the two `final`, closes it cheaply.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -421,11 +552,313 @@ public boolean nextBatch() throws IOException {
return true;
}
+ /**
+ * Splicing emit path. Closes the previous emit's dequeued key vectors
(releasing survivor memory
+ * incrementally), advances to the next row group if needed via {@link
#checkEndOfRowGroup()},
+ * dequeues one survivor key vector per key column, drives non-key column
readers for {@code num}
+ * rows, and assembles a fresh {@link ColumnarBatch} interleaving key
(dequeued) and non-key
+ * (persistent value-vector) slots in the original projection order.
+ * The per-emit batch is a transient view over vectors owned elsewhere; see
{@link #close()} and
+ * {@link #closeSplicingState()}.
+ */
+ private boolean nextBatchSplicing() throws IOException {
Review Comment:
**Design question: why buffer the survivors instead of re-reading the key
columns in phase 2?**
This method, and most of the machinery it depends on, exists only because
phase 2 deliberately omits the key columns. If phase 2 set `requestedSchema`
instead of `nonKeyRequestedSchema` and read under `finalRanges`, the key
columns would come back from the same `PageReadStore`, already aligned with the
non-key ones, and the existing eager `nextBatch()` would serve every row group
unchanged.
That would delete `nextBatchSplicing` (71 lines), `ValueCopier` +
`copierFor` (64), `initializeSplicingState` / `ensureKeyScratchAllocated` /
`ensureCurrentKeyAccumulatorsAllocated` / `appendSurvivorRowToAccumulators` /
`finalizePartialAccumulators` / `closeSplicingState` (~103), the double-close
handling in `close()`, the `skipDataSlots` parameter on `allocateColumns`,
eight fields -- and `ParquetStorageFilter.isSupportedKeyType` with the type
gate it forces into `FileSourceStrategy`.
It also removes, rather than fixes, several things flagged separately in
this review: the whole-row-group buffering, the `LIMIT` stall, the per-row
`byte[]` copy, and the "`keyColumnIndices` must stay sorted" coupling that the
comment below warns can silently swap key columns in the output batch.
The honest cost is re-reading one narrow column's pages under a subset of
the ranges phase 1 already read: always cheaper than phase 1 itself, usually
still in page cache, worst case 2x key decode when the filter passes
everything. That trades IO for the memory cost the design note at L218-224
already concedes, which seems like the easier direction to defend for a first
cut.
If there is a reason this does not work, could it go in the design note? It
is the first thing a reader of this method will ask.
##########
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:
**One throwaway `byte[]` per surviving row, in the innermost loop, for the
most common key shape.**
`WritableColumnVector.getBinary` (L527) does `arrayData().getBytes(off,
len)` -> `new byte[count]` + `System.arraycopy`; `putByteArray` ->
`appendBytes` then copies again. So every survivor costs one garbage array plus
two copies.
For a 1M-row row group at 50% selectivity with a 20-byte string join key
that is 500K `byte[20]` allocations (~20MB of churn including headers) and
~20MB of redundant copying -- per row group, per key column. Dictionary-encoded
pages skip the first allocation but still pay the expansion copy.
A vector-to-vector `appendBytes(int length, WritableColumnVector src, int
srcOffset)` on `WritableColumnVector` -- one
`System.arraycopy`/`Platform.copyMemory` between the two `arrayData()` buffers,
zero allocation -- removes it. The `ByteBuffer` overload at
`WritableColumnVector.java:449` is the existing precedent.
(This disappears entirely if phase 2 reads the key columns; see the review
comment.)
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1885,6 +1885,21 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED =
+ buildConf("spark.sql.parquet.storageFilterPushdown.enabled")
+ .doc("If true, allows the vectorized Parquet reader to evaluate runtime
storage filters " +
+ "(e.g. bloom filters from join runtime filtering) at the scan level
using late " +
+ "materialization: read key columns first, evaluate the filter per row,
then read data " +
+ "columns restricted to surviving rows. This is a planning-time
decision only: when " +
+ "false, no storage filter is attached to a scan in the first place and
the filter is " +
+ "applied as an ordinary post-scan filter instead. Note that the
surviving key values of " +
+ "a whole row group are buffered before the first batch of that row
group is produced, " +
+ "so a task holds up to one extra copy of the key columns for one row
group.")
+ .version("5.0.0")
Review Comment:
Should this be `4.4.0`?
`dev/next_version_candidates.py` currently prints `master 5.0.0 / branch-4.x
4.4.0`, and `branch-4.x` is at `4.4.0-SNAPSHOT` and actively taking backports.
This is an additive, opt-in, default-`false` feature, and the one public type
whose signature changed (`FileSourceScanExec`, a new defaulted parameter) is
under `org.apache.spark.sql.execution.*`, which
`project/MimaExcludes.scala:171` excludes permanently -- so neither master-only
exception (binary-incompatible, or a non-critical dependency bump) seems to
apply.
Unless you are deliberately planning this as master-only, `5.0.0` will make
`docs/sql-configuration` and `SHOW` claim the config arrived a full major later
than it did.
##########
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);
Review Comment:
**This is where `LIMIT` short-circuiting is lost.**
`evaluateStorageFilter` walks the whole row group before this method
returns, so `SELECT ... LIMIT 5` must decode and buffer every key value of row
group 0 -- 1M+ values at the default `parquet.block.size` -- before the first
batch can be produced. The eager path returns after `capacity` rows.
The early-termination tests (`ParquetStorageFilterSuite.scala:1799`,
`:1818`) all use `rowGroupSize = 256L`, so this is not currently exercised.
Chunk-at-a-time evaluation would fix it together with the buffering above, at
the cost of building `finalRanges` incrementally.
--
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]
