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

[rionmonster]

Description

This pull request implements page-level predicate push-down using the Parquet Column Index (#118). This is the page-level equivalent of the row-group-level filtering introduced in #59 within surviving row groups, pages whose per-page min/max statistics prove no rows can match are skipped entirely, avoiding unnecessary decompression and decoding.

The Column Index metadata infrastructure was already in place from PR #116 (e.g., pre-fetched via RowGroupIndexBuffers, parsed via ColumnIndexReader), but wasn't being consumed for filtering, which this wires up.

Additionally, when filtering is active, only the matching pages are fetched from the file instead of the entire column chunk. This can dramatically reduces network I/O on remote backends like S3 for selective filters, using the existing PageRange.coalesce() infrastructure introduced in #31.

Key Changes

The changes related to this effort fall into two categories: net new classes to support the necessary pushdown functionality and extensions to existing classes to take advantage.

New Classes

• Added a new RowRanges class which represents a sorted, non-overlapping interval set representing matching row ranges.
  • Supports intersect() (for AND predicates), union() (for OR), and overlapsPage() for checking if a page's row range intersects any matching interval.
• Added a PageFilterEvaluator** to evaluate an existing FilterPredicate against per-page Column Index min/max statistics to produce RowRanges.
  • Reuses the shared canDrop comparison logic from RowGroupFilterEvaluator.
  • Handles all types (int, long, float, double, boolean, binary) and compound predicates (AND/OR/NOT). Falls back conservatively to all rows when Column Index is absent.
• A new JFR-specific **PageFilterEvent, which is emitted per column per row group when pages are skipped, reporting total/kept/skipped counts.

Additional Classes (related to S3 Optimization)

• Added a new PageRange.forColumn() factory that filters pages by RowRanges, coalesces matching pages into minimal ranges, and extends the first range for dictionary prefix.
• Added a new PageScanner constructor accepting PageRangeData for selective I/O mode. Bifurcated page slicing and dictionary parsing for both full-chunk and page-range modes.
• Updated various readers (e.g. ColumnReader, SingleFileRowReader, etc. ) call sites to attempt per-column page-range I/O when filtering is active (falling back to full chunk fetch for columns that don't benefit or lack an offset index).

Updates and Pipeline Wirting

• Updated PageScanner with an optional RowRanges parameter. When present, scanPagesFromIndex() filters page locations via overlapsPage() before slicing page data, so skipped pages are never decompressed or decoded.
• Updated various readers (e.g. ParqueFileReader, ColumnReader, SingleFileRowReader, etc.) to accept an incoming FilterPredicate to the underlying implementations for supporting row-group computations for filtering.

Verification

A large suite of ~100+ unit tests to validate various filtering operations across all supported operators and types (along with a few edge cases) as well as those to cover the newly introduced RowRanges class (to verify edge-cases and operations like union() and intersect()):

A separate new PageFilterBenchmarkTest which generates a 50M-row synthetic Parquet file with sorted id column and small pages then benchmarks filtered vs. unfiltered reads to validate filtering:

  Results:
    Contender                                      Time (ms)            Rows  Records/sec
    -------------------------------------------------------------------------------------
    Unfiltered [1]                                      64.4      50,000,000  776,319,667
    Unfiltered [2]                                      71.8      50,000,000  696,271,354
    Unfiltered [3]                                      53.1      50,000,000  941,419,406
    Unfiltered [4]                                      52.0      50,000,000  962,348,129
    Unfiltered [5]                                      60.8      50,000,000  822,067,582
    Unfiltered [AVG]                                    60.4      50,000,000  827,520,465

    Filtered (id < 10%) [1]                              6.7       5,000,000  751,524,050
    Filtered (id < 10%) [2]                              7.3       5,000,000  688,764,529
    Filtered (id < 10%) [3]                              6.7       5,000,000  742,900,656
    Filtered (id < 10%) [4]                              6.6       5,000,000  753,508,524
    Filtered (id < 10%) [5]                              6.1       5,000,000  820,922,490
    Filtered (id < 10%) [AVG]                            6.7       5,000,000  751,524,050

    Speedup: 9.0x (60 ms → 7 ms)
    Rows: 50,000,000 → 5,000,000 (90.0% filtered out)

[rionmonster]

@gunnarmorling

Here's a really early first pass for #118. There wasn't really any noticeable difference when targeting our typical NYC Taxi data set (I played around with some attempts to filter down all sub-$100 rides), however the newly added test seems to verify that the filtering is working (roughly a 9x speedup, which makes sense as we filter out 90% of the actual values).

NOTE This PR does already include the changes from #119 so that would need to be merged prior to this. Although I'd like to have you look over it to see if it roughly aligns with what your thoughts were.

[gunnarmorling]

@rionmonster, could you rebase this one and force-push, so as to resolve the merge conflicts? We should then just see the actual changes realated to page-level indexing. Thx!

[rionmonster]

@gunnarmorling

Done — I still have it as a draft in case we want to do any iteration on it before swapping it over to an active PR. Things seem to work but a second set of eyes always helps.

[gunnarmorling]

This great, thank you so much, @rionmonster!

Two tiny comments from me inline. It would also be nice to add a test for column indexes against S3.

I've also asked Claude to take a look, sharing the result for your consideration here. I don't think everything makes sense, so apply what you feel is useful. Also we can talk about how to handle this in general, is it useful to paste such a wall of text here or not. I'm not sure where I stand.

---

Here's the review of PR #143 — page-level Column Index pushdown:

Summary

Well-structured PR. The design cleanly separates concerns: RowRanges as sorted interval math, PageFilterEvaluator as
predicate-to-ranges conversion, and PageScanner as the consumer that skips pages. The reuse of canDrop logic from
RowGroupFilterEvaluator is the right call. The 9x benchmark speedup is convincing.

Must fix

1. Missing tests for And/Or/Not evaluation in PageFilterEvaluator. All tests exercise leaf predicates directly. There
   are no tests for the evaluate() method handling compound predicates — the intersection/union wiring is untested at the
   unit level. RowRangesTest covers intersect/union on RowRanges itself, but the code path that calls them from
   evaluate() isn't exercised.

2. PageFilterBenchmarkTest has no assertion on correctness. If the filter were silently not applied, the test would
   still pass. Add something like assertTrue(filteredRows < unfilteredRows).

Should fix

3. Early exit in overlapsPage(). Since intervals are sorted, once pageLastRow <= ranges[i], no further intervals can
   overlap. Adding this single line avoids scanning the rest of the array for no-match pages:
   if (pageLastRow <= ranges[i]) return false;

4. DRY: column index resolution is duplicated. The pattern of findColumnIndex → fallback to findColumnIndexByPath →
   bounds check appears identically in both RowGroupFilterEvaluator.findStatistics() and
   PageFilterEvaluator.evaluateLeaf(). Extract into a shared helper.

5. union() sorts when both inputs are already sorted. The current implementation creates List<long[]> of individual
   two-element arrays, sorts them, then merges. Since both this.ranges and other.ranges are already sorted and
   non-overlapping, a single-pass merge + coalesce is O(n+m) and avoids the sort and small-array allocations.

6. Boolean type is not tested in PageFilterEvaluatorTest, unlike all other types which have parameterized coverage.

7. Empty RowRanges edge cases. No explicit tests for intersect/union where one side has zero intervals.

Nice to have

8. Long.MAX_VALUE sentinel for last page end. In filterPageLocations(), the last page uses Long.MAX_VALUE as its end
   boundary, while fromPages() correctly uses rowGroupRowCount. The inconsistency is a code smell — consider passing the
   actual row count through.

9. Allocations in intersect()/union(). Both create List<long[]> with many small arrays, then flatten() copies into a
   final long[]. A pre-sized long[] with an index pointer would avoid the intermediate allocations. Minor for typical
   workloads (few intervals), but relevant per CLAUDE.md's "avoid object access and boxing" guideline.

10. Redundant Column Index parsing. For a predicate like AND(col > 5, col < 100), the same column's indexes are parsed
    twice in evaluateLeaf(). Caching parsed indexes per column within a single computeMatchingRows() call would avoid
    this.

11. intersect() clarity. When both intervals end at the same position, only one pointer advances. The current code
    converges correctly but does a wasted iteration. Advancing both pointers on the equal case makes the intent clearer.

[rionmonster]

@gunnarmorling

Thanks for all the feedback! I’ve addressed the items in the must-fix and should-fix sections, along with the allocation-related suggestions (in line with our ongoing efforts to reduce them). I also added an S3-based test for the filtering.

This should be in a pretty good shape after all of these though! Feel free to take a look/merge if it looks good on your end!

[gunnarmorling]

This is great, @rionmonster! One more suggestion for your consideration inline in regards to S3 testing. And a general question: are we coalescing requests for pages which are either directly adjacent or which just have a small gap in between them? This is what we do for row-based pruning already.

[rionmonster]

@gunnarmorling

This is great, @rionmonster! One more suggestion for your consideration inline in regards to S3 testing. And a general question: are we coalescing requests for pages which are either directly adjacent or which just have a small gap in between them? This is what we do for row-based pruning already.

I think this is tied closely to the previous comment regarding the S3 optimization and ensuring we are only reading/downloading the meaningful chunks. Once we incorporate that, we can further optimize downloading adjacent or near-adjacent pages for efficiency.

Going to play around with that a bit as it's a pretty meaningful improvement outside of the benefits from the decoding-level filtering/IO.

[rionmonster]

@gunnarmorling

Okay — it looks like this worked as expected (detecting a filter and only requesting the specific chunks which for the example test case netted an 88% reduction in network bytes downloaded from S3). Pretty huge improvement there. I'll try and update the PR description for historical purposes as this covers that future S3-optimization.

[gunnarmorling]

Noice! Quick heads-up that I've just merged #171, which replaces the S3 SDK (which has quite a heavy dependency footprint) with more light-weight implementation (API requests via the JDK's own HTTP client, request signing via custom implementation of AWS' signing algo, only for authentication against the default credentials provider chain, a new optional module with a much more limited dependency set has been added). This also changes how S3 is accessed in tests, so there's a merge conflict which needs addressing. Thanks a lot, @rionmonster!

[rionmonster]

@gunnarmorling

Cool! I’ll pull those in when I get some time this evening and after that this should be good to go! 👍

[gunnarmorling]

Aaand merged 🥳! Thanks a lot, @rionmonster, great work!