[SPARK-59620][SQL] Late-materialization storage-filter pushdown in the vectorized Parquet reader

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -1947,6 +1947,51 @@ object SQLConf {
       .checkValue(threshold => threshold >= 0, "The threshold must not be negative.")
       .createWithDefault(10)
 
+  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 alone. A pushed filter stays in the post-scan " +
+        "filter as well, the way a pushed data filter does, so honoring it is optional: a reader " +
+        "that meets a row group it cannot prune reads it the way a plain scan would, plus one " +
+        "more read of the key columns, since the phase that evaluated the filter already read " +
+        "them. A file written with no Parquet page index is read with the filter applied only " +
+        "where it empties a whole row group, since narrowing to part of one needs that index, so " +
+        "every row group of it that keeps a row pays that. Setting " +
+        "parquet.filter.columnindex.enabled to false turns this off entirely, because reading " +
+        "part of a row group goes through the page index. 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")
+      .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+      .booleanConf
+      .createWithDefault(false)
+
+  val PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES =
+    buildConf("spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes")
+      .internal()
+      .doc("Most memory, in bytes, that the vectorized Parquet reader holds for one row group " +
+        "while it applies a storage filter. Two things count against it, they sit in different " +
+        "pools, and both grow with the number of surviving rows: the key values buffered to " +
+        "splice into the output batches, which follow the reader's memory mode and so can be off " +
+        "heap, and the row ranges those rows fall into, always on heap, which the second phase " +
+        "needs to select its pages. The count is examined after every surviving row. Past the " +
+        "limit the reader releases the buffer and reads every projected column of the surviving " +
+        "rows instead, " +
+        "which costs one extra read of the key columns, and past it again it reads the row group " +
+        "with no filter applied at all, which is correct but as slow as not pushing the filter. " +
+        "What is counted is the buffered values and their per-row overhead, not the backing " +
+        "arrays, which a column vector may grow beyond that.")
+      .version("5.0.0")
+      .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+      .bytesConf(ByteUnit.BYTE)
+      .checkValue(_ > 0, "must be positive")
+      .createWithDefaultString("64MB")
+
   val PARQUET_AGGREGATE_PUSHDOWN_ENABLED = buildConf("spark.sql.parquet.aggregatePushdown")
     .doc("If true, aggregates will be pushed down to Parquet for optimization. Support MIN, MAX " +
       "and COUNT as aggregate expression. For MIN/MAX, support boolean, integer, float and date " +
@@ -9118,6 +9163,12 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
   def parquetFilterPushDownInFilterThreshold: Int =
     getConf(PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD)
 
+  def parquetStorageFilterPushdownEnabled: Boolean =
+    getConf(PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED)
+
+  def parquetStorageFilterPushdownMaxSplicedRowGroupBytes: Long =
+    getConf(PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES)
+
   def parquetAggregatePushDown: Boolean = getConf(PARQUET_AGGREGATE_PUSHDOWN_ENABLED)
 
   def orcFilterPushDown: Boolean = getConf(ORC_FILTER_PUSHDOWN_ENABLED)

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java

@@ -19,30 +19,29 @@
 
 import org.apache.parquet.column.ColumnDescriptor;
 
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
 import java.util.PrimitiveIterator;
 
 /**
  * Helper class to store intermediate state while reading a Parquet column chunk.
  */
 final class ParquetReadState {
-  /** A special row range used when there is no row indexes (hence all rows must be included) */
-  private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+  /** The row indexes to include, only not-null if the column index is present. */
+  private final PrimitiveIterator.OfLong rowIndexes;
 
   /**
-   * A special row range used when the row indexes are present AND all the row ranges have been
-   * processed. This serves as a sentinel at the end indicating that all rows come after the last
-   * row range should be skipped.
+   * The current row range, as its bounds rather than as an object: one range per surviving row is
+   * what a filter with scattered survivors produces, for every column reader of every row group.
+   *
+   * <p>With no row indexes they are the whole range, since every row must be included. Once the
+   * indexes are exhausted they are inverted, which says that every row from there on is to be
+   * skipped.
    */
-  private static final RowRange END_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+  private long currentRangeStart;
+  private long currentRangeEnd;
 
-  /** Iterator over all row ranges, only not-null if column index is present */
-  private final Iterator<RowRange> rowRanges;
-
-  /** The current row range */
-  private RowRange currentRange;
+  /** The row index that ended the current range by not continuing it, so it starts the next one. */
+  private long pendingRowIndex;
+  private boolean hasPendingRowIndex;
 
   /** Maximum repetition level for the Parquet column */
   final int maxRepetitionLevel;
@@ -90,43 +89,10 @@ final class ParquetReadState {
     this.maxRepetitionLevel = descriptor.getMaxRepetitionLevel();
     this.maxDefinitionLevel = descriptor.getMaxDefinitionLevel();
     this.isRequired = isRequired;
-    this.rowRanges = constructRanges(rowIndexes);
+    this.rowIndexes = rowIndexes;
     nextRange();
   }
 
-  /**
-   * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
-   * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
-   * `[0-2], [4-5], [7-9]`.
-   */
-  private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
-    if (rowIndexes == null) {
-      return null;
-    }
-
-    List<RowRange> rowRanges = new ArrayList<>();
-    long currentStart = Long.MIN_VALUE;
-    long previous = Long.MIN_VALUE;
-
-    while (rowIndexes.hasNext()) {
-      long idx = rowIndexes.nextLong();
-      if (currentStart == Long.MIN_VALUE) {
-        currentStart = idx;
-      } else if (previous + 1 != idx) {
-        RowRange range = new RowRange(currentStart, previous);
-        rowRanges.add(range);
-        currentStart = idx;
-      }
-      previous = idx;
-    }
-
-    if (previous != Long.MIN_VALUE) {
-      rowRanges.add(new RowRange(currentStart, previous));
-    }
-
-    return rowRanges.iterator();
-  }
-
   /**
    * Must be called at the beginning of reading a new batch.
    */
@@ -151,32 +117,51 @@ void resetForNewPage(int totalValuesInPage, long pageFirstRowIndex) {
    * Returns the start index of the current row range.
    */
   long currentRangeStart() {
-    return currentRange.start;
+    return currentRangeStart;
   }
 
   /**
    * Returns the end index of the current row range.
    */
   long currentRangeEnd() {
-    return currentRange.end;
+    return currentRangeEnd;
   }
 
   /**
-   * Advance to the next range.
+   * Advances to the next range, coalescing the run of ascending row indexes that forms it. For
+   * example `[0, 1, 2, 4, 5, 7, 8, 9]` yields `[0-2]`, then `[4-5]`, then `[7-9]`.
+   *
+   * <p>One range at a time on purpose. They are consumed once, in order, so holding them all buys
+   * nothing and costs a list per column reader of the row group, which for scattered survivors is
+   * one entry per row in every one of those lists.
    */
   void nextRange() {
-    if (rowRanges == null) {
-      currentRange = MAX_ROW_RANGE;
-    } else if (!rowRanges.hasNext()) {
-      currentRange = END_ROW_RANGE;
-    } else {
-      currentRange = rowRanges.next();
+    if (rowIndexes == null) {
+      currentRangeStart = Long.MIN_VALUE;
+      currentRangeEnd = Long.MAX_VALUE;
+      return;
     }
-  }
-
-  /**
-   * Helper struct to represent a range of row indexes `[start, end]`.
-   */
-  private record RowRange(long start, long end) {
+    if (!hasPendingRowIndex && !rowIndexes.hasNext()) {
+      currentRangeStart = Long.MAX_VALUE;
+      currentRangeEnd = Long.MIN_VALUE;
+      return;
+    }
+    long start = hasPendingRowIndex ? pendingRowIndex : rowIndexes.nextLong();
+    hasPendingRowIndex = false;
+    long end = start;
+    // A range can only be closed by seeing the index that does not continue it, so that index is
+    // held back for the next call.
+    while (rowIndexes.hasNext()) {
+      long idx = rowIndexes.nextLong();
+      if (idx == end + 1) {
+        end = idx;
+      } else {
+        pendingRowIndex = idx;
+        hasPendingRowIndex = true;
+        break;
+      }
+    }
+    currentRangeStart = start;
+    currentRangeEnd = end;
   }
 }

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/SpecificParquetRecordReaderBase.java

@@ -278,6 +278,16 @@ interface ParquetRowGroupReader extends Closeable {
      * Reads the next row group from this reader. Returns null if there is no more row group.
      */
     PageReadStore readNextRowGroup() throws IOException;
+
+    /**
+     * Returns the underlying {@link ParquetFileReader}, or null if this reader does not wrap one
+     * (e.g. test implementations). Callers can use this to access lower-level APIs such as
+     * {@code setRequestedSchema}, {@code readRowGroup(int)} and
+     * {@code readFilteredRowGroup(int, RowRanges)} which are needed for late materialization.
+     */
+    default ParquetFileReader getUnderlyingReader() {
+      return null;
+    }
   }
 
   private static class ParquetRowGroupReaderImpl implements ParquetRowGroupReader {
@@ -292,6 +302,11 @@ public PageReadStore readNextRowGroup() throws IOException {
       return reader.readNextFilteredRowGroup();
     }
 
+    @Override
+    public ParquetFileReader getUnderlyingReader() {
+      return reader;
+    }
+
     @Override
     public void close() throws IOException {
       if (reader != null) {