#118 Support for Page-Level Column Index Pushdown/Filtering

README.md

@@ -362,7 +362,7 @@ while (rowReader.hasNext()) {
 
 ### Predicate Pushdown (Filter)
 
-Filter predicates allow Hardwood to skip entire row groups whose statistics prove that no rows can match the predicate, avoiding unnecessary I/O and decoding.
+Filter predicates enable two levels of predicate pushdown. At the row-group level, entire row groups whose statistics prove no rows can match are skipped. Within surviving row groups, the Column Index (per-page min/max statistics) is used to skip individual pages, avoiding unnecessary decompression and decoding. On remote backends like S3, only the matching pages are fetched, reducing network I/O.
 
 ```java
 import dev.hardwood.reader.FilterPredicate;
@@ -836,12 +836,13 @@ Or attach dynamically via `jcmd <pid> JFR.start`.
 | `dev.hardwood.RowGroupScanned` | Decode | Page boundaries scanned in a column chunk. Fields: file, rowGroupIndex, column, pageCount, scanStrategy (`sequential` or `offset-index`) |
 | `dev.hardwood.PageDecoded` | Decode | Single data page decoded. Fields: column, compressedSize, uncompressedSize |
 | `dev.hardwood.RowGroupFilter` | Filter | Row groups filtered by predicate pushdown. Fields: file, totalRowGroups, rowGroupsKept, rowGroupsSkipped |
+| `dev.hardwood.PageFilter` | Filter | Pages filtered by Column Index predicate pushdown. Fields: file, rowGroupIndex, column, totalPages, pagesKept, pagesSkipped |
 | `dev.hardwood.BatchWait` | Pipeline | Consumer blocked waiting for the assembly pipeline. Fields: column |
 | `dev.hardwood.PrefetchMiss` | Pipeline | Prefetch queue miss requiring synchronous decode. Fields: file, column, newDepth, queueEmpty |
 
 Events appear under the **Hardwood** category in JDK Mission Control (JMC) or any JFR analysis tool. Use them to identify:
 - **I/O bottlenecks** — large `FileMapping` durations or frequent `PrefetchMiss` events
-- **Filter effectiveness** — `RowGroupFilter` shows how many row groups were skipped
+- **Filter effectiveness** — `RowGroupFilter` shows how many row groups were skipped; `PageFilter` shows how many pages were skipped within surviving row groups
 - **Decode hotspots** — `PageDecoded` events with large uncompressed sizes or high frequency
 - **Pipeline stalls** — `BatchWait` events indicate the reader is waiting for decoded data
 

core/src/main/java/dev/hardwood/internal/reader/FileManager.java

@@ -16,10 +16,13 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 import dev.hardwood.InputFile;
+import dev.hardwood.internal.thrift.OffsetIndexReader;
+import dev.hardwood.internal.thrift.ThriftCompactReader;
 import dev.hardwood.jfr.FileOpenedEvent;
 import dev.hardwood.jfr.RowGroupFilterEvent;
 import dev.hardwood.metadata.ColumnChunk;
 import dev.hardwood.metadata.FileMetaData;
+import dev.hardwood.metadata.OffsetIndex;
 import dev.hardwood.metadata.PhysicalType;
 import dev.hardwood.metadata.RowGroup;
 import dev.hardwood.reader.FilterPredicate;
@@ -331,33 +334,85 @@ private List<List<PageInfo>> scanAllProjectedColumns(InputFile inputFile, Opened
                 throw new UncheckedIOException("Failed to fetch index buffers for row group " + rgIdx, e);
             }
 
-            List<ChunkRange> chunkRanges = ChunkRange.coalesce(
-                    rowGroup.columns(), columnIndices, ChunkRange.MAX_GAP_BYTES);
-            ByteBuffer[] rangeBuffers = new ByteBuffer[chunkRanges.size()];
-            try {
-                for (int r = 0; r < chunkRanges.size(); r++) {
-                    ChunkRange range = chunkRanges.get(r);
-                    rangeBuffers[r] = inputFile.readRange(range.offset(), range.length());
+            // Compute matching row ranges for page-level Column Index filtering
+            RowRanges matchingRows = RowRanges.ALL;
+            if (filterPredicate != null) {
+                matchingRows = PageFilterEvaluator.computeMatchingRows(
+                        filterPredicate, rowGroup, openedFile.schema, indexBuffers);
+            }
+
+            // Try page-range I/O for each column when filtering is active.
+            // Columns that don't benefit fall back to full chunk fetching.
+            final RowRanges rowRanges = matchingRows;
+            PageRangeData[] pageRangeDataPerColumn = new PageRangeData[projectedColumnCount];
+
+            if (!matchingRows.isAll()) {
+                for (int projectedIndex = 0; projectedIndex < projectedColumnCount; projectedIndex++) {
+                    int columnIndex = columnIndices[projectedIndex];
+                    ColumnIndexBuffers colBuffers = indexBuffers.forColumn(columnIndex);
+                    if (colBuffers != null && colBuffers.offsetIndex() != null) {
+                        try {
+                            OffsetIndex offsetIndex = OffsetIndexReader.read(
+                                    new ThriftCompactReader(colBuffers.offsetIndex()));
+                            ColumnChunk columnChunk = rowGroup.columns().get(columnIndex);
+                            List<PageRange> pageRanges = PageRange.forColumn(
+                                    offsetIndex, matchingRows, columnChunk, rowGroup.numRows(), ChunkRange.MAX_GAP_BYTES);
+                            if (!pageRanges.isEmpty()) {
+                                pageRangeDataPerColumn[projectedIndex] = PageRangeData.fetch(inputFile, pageRanges);
+                            }
+                        }
+                        catch (IOException e) {
+                            throw new UncheckedIOException("Failed to fetch page ranges for row group " + rgIdx, e);
+                        }
+                    }
                 }
             }
-            catch (IOException e) {
-                throw new UncheckedIOException("Failed to fetch column chunk data for row group " + rgIdx, e);
+
+            // Fetch full chunks for columns not using page-range I/O
+            List<ChunkRange> chunkRanges = null;
+            ByteBuffer[] rangeBuffers = null;
+            for (int projectedIndex = 0; projectedIndex < projectedColumnCount; projectedIndex++) {
+                if (pageRangeDataPerColumn[projectedIndex] == null) {
+                    // At least one column needs full chunk data — fetch all chunks once
+                    chunkRanges = ChunkRange.coalesce(
+                            rowGroup.columns(), columnIndices, ChunkRange.MAX_GAP_BYTES);
+                    rangeBuffers = new ByteBuffer[chunkRanges.size()];
+                    try {
+                        for (int r = 0; r < chunkRanges.size(); r++) {
+                            ChunkRange range = chunkRanges.get(r);
+                            rangeBuffers[r] = inputFile.readRange(range.offset(), range.length());
+                        }
+                    }
+                    catch (IOException e) {
+                        throw new UncheckedIOException("Failed to fetch column chunk data for row group " + rgIdx, e);
+                    }
+                    break;
+                }
             }
 
             for (int projectedIndex = 0; projectedIndex < projectedColumnCount; projectedIndex++) {
                 final int columnIndex = columnIndices[projectedIndex];
                 final ColumnSchema columnSchema = columnSchemas[projectedIndex];
                 final ColumnChunk columnChunk = rowGroup.columns().get(columnIndex);
+                final PageRangeData colPageRangeData = pageRangeDataPerColumn[projectedIndex];
 
-                final ByteBuffer colChunkData = sliceColumnChunk(chunkRanges, rangeBuffers, columnChunk);
-                final long colChunkOffset = chunkStartOffset(columnChunk);
+                final PageScanner scanner;
+                if (colPageRangeData != null) {
+                    scanner = new PageScanner(columnSchema, columnChunk, context,
+                            colPageRangeData, indexBuffers.forColumn(columnIndex),
+                            rgIdx, fileName, rowRanges);
+                }
+                else {
+                    final ByteBuffer colChunkData = sliceColumnChunk(chunkRanges, rangeBuffers, columnChunk);
+                    final long colChunkOffset = chunkStartOffset(columnChunk);
+                    scanner = new PageScanner(columnSchema, columnChunk, context,
+                            colChunkData, colChunkOffset, indexBuffers.forColumn(columnIndex),
+                            rgIdx, fileName, rowRanges);
+                }
 
                 final CompletableFuture<List<PageInfo>> previousFuture = scanFutures[projectedIndex];
                 scanFutures[projectedIndex] = previousFuture.thenCombineAsync(
                         CompletableFuture.supplyAsync(() -> {
-                            PageScanner scanner = new PageScanner(columnSchema, columnChunk, context,
-                                    colChunkData, colChunkOffset, indexBuffers.forColumn(columnIndex),
-                                    rgIdx, fileName);
                             try {
                                 return scanner.scanPages();
                             }

core/src/main/java/dev/hardwood/internal/reader/PageFilterEvaluator.java

@@ -0,0 +1,174 @@
+/*
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Copyright The original authors
+ *
+ *  Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package dev.hardwood.internal.reader;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.List;
+
+import dev.hardwood.internal.thrift.ColumnIndexReader;
+import dev.hardwood.internal.thrift.OffsetIndexReader;
+import dev.hardwood.internal.thrift.ThriftCompactReader;
+import dev.hardwood.metadata.ColumnIndex;
+import dev.hardwood.metadata.OffsetIndex;
+import dev.hardwood.metadata.PageLocation;
+import dev.hardwood.metadata.RowGroup;
+import dev.hardwood.reader.FilterPredicate;
+import dev.hardwood.reader.FilterPredicate.And;
+import dev.hardwood.reader.FilterPredicate.BinaryColumnPredicate;
+import dev.hardwood.reader.FilterPredicate.BooleanColumnPredicate;
+import dev.hardwood.reader.FilterPredicate.DoubleColumnPredicate;
+import dev.hardwood.reader.FilterPredicate.FloatColumnPredicate;
+import dev.hardwood.reader.FilterPredicate.IntColumnPredicate;
+import dev.hardwood.reader.FilterPredicate.LongColumnPredicate;
+import dev.hardwood.reader.FilterPredicate.Not;
+import dev.hardwood.reader.FilterPredicate.Or;
+import dev.hardwood.schema.FileSchema;
+
+/// Evaluates a [FilterPredicate] against per-page statistics from the Column Index
+/// to produce [RowRanges] representing rows that might match.
+///
+/// This is the page-level equivalent of [RowGroupFilterEvaluator]. While that class
+/// decides whether an entire row group can be skipped, this class determines which
+/// pages within a surviving row group can be skipped.
+public class PageFilterEvaluator {
+
+    /// Computes the row ranges within a row group that might match the given predicate,
+    /// based on per-page min/max statistics from the Column Index.
+    ///
+    /// Returns `RowRanges.all()` when the Column Index is absent or the predicate
+    /// cannot be evaluated at the page level (conservative fallback).
+    ///
+    /// @param predicate    the filter predicate to evaluate
+    /// @param rowGroup     the row group to evaluate against
+    /// @param schema       the file schema for column resolution
+    /// @param indexBuffers pre-fetched index buffers for the row group
+    /// @return row ranges that might contain matching rows
+    public static RowRanges computeMatchingRows(FilterPredicate predicate, RowGroup rowGroup,
+            FileSchema schema, RowGroupIndexBuffers indexBuffers) {
+        long rowCount = rowGroup.numRows();
+        return evaluate(predicate, rowGroup, schema, indexBuffers, rowCount);
+    }
+
+    private static RowRanges evaluate(FilterPredicate predicate, RowGroup rowGroup,
+            FileSchema schema, RowGroupIndexBuffers indexBuffers, long rowCount) {
+        return switch (predicate) {
+            case IntColumnPredicate p -> evaluateLeaf(p.column(), rowGroup, schema, indexBuffers, rowCount,
+                    (ci, i) -> {
+                        int min = StatisticsDecoder.decodeInt(ci.minValues().get(i));
+                        int max = StatisticsDecoder.decodeInt(ci.maxValues().get(i));
+                        return RowGroupFilterEvaluator.canDrop(p.op(), p.value(), min, max);
+                    });
+            case LongColumnPredicate p -> evaluateLeaf(p.column(), rowGroup, schema, indexBuffers, rowCount,
+                    (ci, i) -> {
+                        long min = StatisticsDecoder.decodeLong(ci.minValues().get(i));
+                        long max = StatisticsDecoder.decodeLong(ci.maxValues().get(i));
+                        return RowGroupFilterEvaluator.canDrop(p.op(), p.value(), min, max);
+                    });
+            case FloatColumnPredicate p -> evaluateLeaf(p.column(), rowGroup, schema, indexBuffers, rowCount,
+                    (ci, i) -> {
+                        float min = StatisticsDecoder.decodeFloat(ci.minValues().get(i));
+                        float max = StatisticsDecoder.decodeFloat(ci.maxValues().get(i));
+                        return RowGroupFilterEvaluator.canDropFloat(p.op(), p.value(), min, max);
+                    });
+            case DoubleColumnPredicate p -> evaluateLeaf(p.column(), rowGroup, schema, indexBuffers, rowCount,
+                    (ci, i) -> {
+                        double min = StatisticsDecoder.decodeDouble(ci.minValues().get(i));
+                        double max = StatisticsDecoder.decodeDouble(ci.maxValues().get(i));
+                        return RowGroupFilterEvaluator.canDropDouble(p.op(), p.value(), min, max);
+                    });
+            case BooleanColumnPredicate p -> evaluateLeaf(p.column(), rowGroup, schema, indexBuffers, rowCount,
+                    (ci, i) -> {
+                        int min = StatisticsDecoder.decodeBoolean(ci.minValues().get(i)) ? 1 : 0;
+                        int max = StatisticsDecoder.decodeBoolean(ci.maxValues().get(i)) ? 1 : 0;
+                        int value = p.value() ? 1 : 0;
+                        return RowGroupFilterEvaluator.canDrop(p.op(), value, min, max);
+                    });
+            case BinaryColumnPredicate p -> evaluateLeaf(p.column(), rowGroup, schema, indexBuffers, rowCount,
+                    (ci, i) -> {
+                        byte[] min = ci.minValues().get(i);
+                        byte[] max = ci.maxValues().get(i);
+                        int cmpMin = StatisticsDecoder.compareBinary(p.value(), min);
+                        int cmpMax = StatisticsDecoder.compareBinary(p.value(), max);
+                        return RowGroupFilterEvaluator.canDropCompared(p.op(), cmpMin, cmpMax,
+                                StatisticsDecoder.compareBinary(min, max));
+                    });
+            case And a -> {
+                RowRanges result = RowRanges.all(rowCount);
+                for (FilterPredicate child : a.filters()) {
+                    result = result.intersect(evaluate(child, rowGroup, schema, indexBuffers, rowCount));
+                }
+                yield result;
+            }
+            case Or o -> {
+                RowRanges result = null;
+                for (FilterPredicate child : o.filters()) {
+                    RowRanges childRanges = evaluate(child, rowGroup, schema, indexBuffers, rowCount);
+                    result = (result == null) ? childRanges : result.union(childRanges);
+                }
+                yield (result != null) ? result : RowRanges.all(rowCount);
+            }
+            case Not ignored -> RowRanges.all(rowCount);
+        };
+    }
+
+    /// Evaluates a leaf predicate against the Column Index for a single column,
+    /// producing RowRanges for pages that cannot be dropped.
+    private static RowRanges evaluateLeaf(String columnName, RowGroup rowGroup,
+            FileSchema schema, RowGroupIndexBuffers indexBuffers, long rowCount,
+            PageCanDropTest canDropTest) {
+
+        int columnIndex = RowGroupFilterEvaluator.resolveColumnIndex(columnName, rowGroup, schema);
+        if (columnIndex < 0) {
+            return RowRanges.all(rowCount);
+        }
+
+        ColumnIndexBuffers colBuffers = indexBuffers.forColumn(columnIndex);
+        if (colBuffers == null || colBuffers.columnIndex() == null || colBuffers.offsetIndex() == null) {
+            return RowRanges.all(rowCount);
+        }
+
+        ColumnIndex columnIdx;
+        OffsetIndex offsetIdx;
+        try {
+            columnIdx = ColumnIndexReader.read(new ThriftCompactReader(colBuffers.columnIndex()));
+            offsetIdx = OffsetIndexReader.read(new ThriftCompactReader(colBuffers.offsetIndex()));
+        }
+        catch (IOException e) {
+            throw new UncheckedIOException("Failed to parse Column/Offset Index for column '" + columnName + "'", e);
+        }
+
+        return evaluatePages(columnIdx, offsetIdx, rowCount, canDropTest);
+    }
+
+    /// Evaluates a keep bitmap for pages using pre-parsed Column Index and Offset Index.
+    static RowRanges evaluatePages(ColumnIndex columnIdx, OffsetIndex offsetIdx,
+            long rowCount, PageCanDropTest canDropTest) {
+        List<PageLocation> pages = offsetIdx.pageLocations();
+        int pageCount = pages.size();
+        boolean[] keep = new boolean[pageCount];
+
+        for (int i = 0; i < pageCount; i++) {
+            // Null-only pages cannot match any value predicate
+            if (columnIdx.nullPages().get(i)) {
+                continue;
+            }
+            // Keep the page if it cannot be dropped
+            keep[i] = !canDropTest.canDrop(columnIdx, i);
+        }
+
+        return RowRanges.fromPages(pages, keep, rowCount);
+    }
+
+    /// Functional interface for testing whether a page can be dropped based on its
+    /// Column Index min/max values.
+    @FunctionalInterface
+    interface PageCanDropTest {
+        boolean canDrop(ColumnIndex columnIndex, int pageIndex);
+    }
+}