What's the recommended way to read parquet files with offset + len

[wooden-worm]

I want to read just a portion of a given parquet file, from row number x to x + n. What's the recommended way to do so?

I've been playing around with read::get_page_stream, and then skip pages based on its size, but not sure that's the optimal way.

[jorgecarleitao]

Great question. :)

We currently do not support offset pushdown in general parquet.

With that said, we do support it on (either) situations:

1. pages are written in V2. In this case, we can use the page header to skip based on the page's number of values. Essentially, create a new page iterator that skips pages up to a number of values.
2. the parquet file has page offsets. This is a feature that allow us to know which page contains which intervals of rows.

Do you know whether you are in one of these situations?

Regardless of this, there is a use-case to add support for offset pushdown to the deserializer.

[wooden-worm]

Hi Jorge, I'm in situation 1, my parquet file is in V2. Can you give an example of creating a new page iterator to skip pages? Is read::get_page_stream the correct way to do so? I'm thinking of getting the page stream, and then do something like

page_stream.skip_while(|page|  { 
... logic to count rows & check whether we should skip this page
}

Does it look correct?

Also, thank you for your awesome work on arrow2 & parquet2

[jorgecarleitao]

Hi @wooden-worm

thank you for your kind words 🙇

Something like this:

use std::sync::Arc;

use futures::pin_mut;
use futures::StreamExt;
use tokio::fs::File;
use tokio::io::BufReader;
use tokio_util::compat::*;

use parquet2::page;
use parquet2::error::Error;
use parquet2::read::{get_page_stream, read_metadata_async};

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Error> {
    use std::env;
    let args: Vec<String> = env::args().collect();
    let file_path = Box::new(args[1].clone());

    // # Read metadata
    let mut reader = BufReader::new(File::open(file_path.as_ref()).await?).compat();

    let metadata = read_metadata_async(&mut reader).await?;

    // metadata
    println!("number of rows: {}", metadata.num_rows);

    // * first row group
    // * first column
    let column_metadata = &metadata.row_groups[0].columns()[0];

    let pages = get_page_stream(
        column_metadata,
        &mut reader,
        vec![],
        Arc::new(|_, _| true),
        usize::MAX,
    )
    .await?;

    pin_mut!(pages); // needed for iteration

    let first_page = pages.next().await.unwrap()?;

    match first_page {
        page::CompressedPage::Data(page) => match page.header() {
            page::DataPageHeader::V1(_) => todo!("v1 not supported for offset pushdown"),
            page::DataPageHeader::V2(header) => {
                // this can be used to skip
                let rows_in_page = header.num_rows;
            },
        },
        page::CompressedPage::Dict(_) => {
            // these we always consume, as dictionaries are needed to get the values of the pages
        },
    };
    Ok(())
}

[wooden-worm]

Thank you Jorge, that answers my question 👍