[alamb]

Which issue does this PR close?

• Part of [EPIC] Improve I/O patterns for Parquet scans #26682
• Follow on to test(parquet): add I/O tracking test for Bloom filter loading baseline #26897

Rationale for this change

We have several issues / PRs open that contemplate changing the I/O pattern such as

• Parquet Pushdown Filters: Way to keep the I/O pattern the same when pushdown filters are enabled #24393
• [EPIC] Fix performance regressions when enabling parquet filter pushdown (late materialization) #20324
• WIP: Add parquet progressive_io option using ParquetPushDecoder::scan_plan (one-shot row group I/O with pushdown filters) #26387
• WIP: Enable parquet pushdown_filters and progressive_io by default #26418
• Prefetch Parquet row groups with a bounded I/O budget #23492
• POC: batch-granular Parquet scans with the push decoder and scan_plan read-ahead #24086
• perf: evaluate Parquet Bloom filters incrementally during row group pruning #25811

I want to have a test that clearly shows how those proposals effect the scan

What changes are included in this PR?

Add some new tests that show the Parquet I/O patterns. For example I think this is quite clear now the difference in I/O between filter pushdown and not

/// Without filter pushdown, the scan fetches every column of every row group
#[tokio::test]
async fn test_filter_pushdown_io_disabled() {
    let test = filter_pushdown_test(false).await;
    assert_snapshot!(test.query("SELECT * FROM t WHERE val = 7").await, @r"
    --- Query: SELECT * FROM t WHERE val = 7 (1 rows) ---
    Request 1
        Read Parquet footer (offsets 0..1402)
    Request 2
        Read dictionary page and data page 0 for column 'id' (rg 0) (offsets 4..51)
        Read dictionary page and data page 0 for column 'val' (rg 0) (offsets 51..98)
    Request 3
        Read dictionary page and data page 0 for column 'id' (rg 1) (offsets 98..145)
        Read dictionary page and data page 0 for column 'val' (rg 1) (offsets 145..192)
    Request 4
        Read dictionary page and data page 0 for column 'id' (rg 2) (offsets 192..239)
        Read dictionary page and data page 0 for column 'val' (rg 2) (offsets 239..286)
    Produce batch (1 rows)
    ");
}

/// **With** filter pushdown, the scan first fetches and evaluates the filter
/// column (`val`) for each row group, and then fetches the remaining columns
/// (`id`) only for the row groups with matching rows (row group 1).
#[tokio::test]
async fn test_filter_pushdown_io_enabled() {
    let test = filter_pushdown_test(true).await;
    assert_snapshot!(test.query("SELECT * FROM t WHERE val = 7").await, @r"
    --- Query: SELECT * FROM t WHERE val = 7 (1 rows) ---
    Request 1
        Read Parquet footer (offsets 0..1402)
    Request 2
        Read dictionary page and data page 0 for column 'val' (rg 0) (offsets 51..98)
    Request 3
        Read dictionary page and data page 0 for column 'val' (rg 1) (offsets 145..192)
    Request 4
        Read dictionary page for column 'id' (rg 1) (offsets 98..124)
        Read data page 0 for column 'id' (rg 1) (offsets 124..145)
    Produce batch (1 rows)
    Request 5
        Read dictionary page and data page 0 for column 'val' (rg 2) (offsets 239..286)
    ");
}

What is the testing strategy for this PR?

CI

Are there any user-facing changes?

No

[alamb]

@adriangb @Dandandan @zhuqi-lucas and @xudong963 -- I wonder what you think about this style test to illustrate the IO patterns in our existing parquet reader?

I am hoping that as we improve the patterns we'll be able to use this style test to more easily document and verify the behavior

[adriangb]

I like the idea a lot!

I wonder if there's isn't a more general version of this, e.g. if we added tracing as a dev-dep and sprinkled tracing::debug! around in important places so we could track planning and other operations as well.

[zhuqi-lucas]

Nice harness, this makes the I/O differences very easy to see. A few small things, none blocking:

1. The test files are small enough that "Read Parquet footer" is the whole file (offsets 0..1402 / 0..838). In page_pruning_io the page index is clearly used (only page 1 is read for id = 5) but never shows up as its own request, while bloom_filter_io shows the OffsetIndex as a separate one. A short note (or a file where the footer read does not cover everything) would help readers not assume the page index is free.
2. describe_range returns on the first overlapping column chunk, so a coalesced request spanning several chunks would only show the first one.
3. record_request and record_range take the lock separately, so with more than one partition the Request N lines could interleave with another request's reads. Fine with target_partitions = 1 today.

Nits: "acess" in the title, and double spaces in the topk_io.rs docs ("TopK filter is", "which proves").

[alamb]

I do like ascii art

[alamb]

I think showing the distinct requests is what is important as that is the actual IO that is generated

[alamb]

I do love seeing this difference

[alamb]

this is infrastructure to handle tracking what byte ranges mean

[alamb]

I like the idea a lot!

I wonder if there's isn't a more general version of this, e.g. if we added tracing as a dev-dep and sprinkled tracing::debug! around in important places so we could track planning and other operations as well.

I reviewed

• prototype: capture Parquet scan operations with tracing in I/O tests #27110

It is a great idea -- I don't like the use of the tracing crate, but otherwise 👌

[alamb]

1. The test files are small enough that "Read Parquet footer" is the whole file (offsets 0..1402 / 0..838). In page_pruning_io the page index is clearly used (only page 1 is read for id = 5) but never shows up as its own request, while bloom_filter_io shows the OffsetIndex as a separate one. A short note (or a file where the footer read does not cover everything) would help readers not assume the page index is free.

This is an excellent point -- I actually adjusted the test so that it sets the metadata hint so that it only covers the metadata and not also the bloom filters and offset indexes

2. describe_range returns on the first overlapping column chunk, so a coalesced request spanning several chunks would only show the first one.

Fixex

3. record_request and record_range take the lock separately, so with more than one partition the Request N lines could interleave with another request's reads. Fine with target_partitions = 1 today.

Fixex

Nits: "acess" in the title, and double spaces in the topk_io.rs docs ("TopK filter is", "which proves").

Fixed

[alamb]

Thanks everyone

[alamb]

I'll merge this in and hopefully we can keep iterating to make the tests easier to understand / cover more