apache / apache/arrow-rs

Multithreaded Reading Parquet from `std::fs::File`

Open
#7,667 1 comment 0 reactions 0 assignees View on GitHub
documentation question
Dominant language
Rust
Stars
3.6k
Forks
1.3k
Avg merge
2d 18h
Merged PRs (30d)
169

Description

**Which part is this question about**
library api: multithreaded reading without converting to arrow

**Describe your question**
I am trying to read a single file from multiple threads. The `ColumnReader`s return lots of protocol errors and sometimes panics. I am not sure if this is a bug or if I am using the library wrong. If the latter, I'd love to know what the correct way is.

**Additional context**
I am guessing that the issue is that multiple `RowGroupReader`s share the same file handle and interfere with each other by seeking. However, `SerializedFileReader::get_row_group` seems to be inviting me to do exactly this kind of sharing by taking a shared reference to the `SerializedFileReader`.

In my example, I use the lineitem table from TPC-H, generated using [tpchgen-cli](https://github.com/clflushopt/tpchgen-rs):
```sh
mkdir tpch-data
cd tpch-data
tpchgen-cli -s 1 --format=parquet
cd ..
```

Here is the code:
```rust
use parquet::column::reader::ColumnReader;
use parquet::file::metadata::RowGroupMetaData;
use parquet::file::reader::{FileReader, RowGroupReader, SerializedFileReader};
use rayon::prelude::*;

fn find_col(metadata: &RowGroupMetaData, reader: &dyn RowGroupReader, name: &str) -> ColumnReader {
for (i, x) in metadata.columns().iter().enumerate() {
if x.column_descr().name() == name {
return reader.get_column_reader(i).unwrap();
}
}
panic!("column {name:?} not found");
}

fn main() {
let reader =
SerializedFileReader::new(std::fs::File::open("./tpch-data/lineitem.parquet").unwrap())
.unwrap();
let metadata = reader.metadata();
(0..metadata.num_row_groups())
.into_par_iter()
.for_each(|i| {
let metadata = &metadata.row_group(i);
let reader = reader.get_row_group(i).unwrap();
let ColumnReader::Int64ColumnReader(mut reader_l_quantity_112) =
find_col(metadata, &*reader, "l_quantity")
else {
panic!()
};
let mut read_buffer_l_quantity_113 = Vec::new();
loop {
let read_count_126 = reader_l_quantity_112
.read_records(10000, None, None, &mut read_buffer_l_quantity_113)
.unwrap()
.0;
if read_count_126 == 0 {
break;
}
}
})
}
```

Here are some of the errors I am seeing:
```
thread '' panicked at src/bin/parquet_issue.rs:34:22:
called `Result::unwrap()` on an `Err` value: External(ProtocolError { kind: Unknown, message: "cannot skip field type Stop" })

thread '' panicked at src/bin/parquet_issue.rs:34:22:
called `Result::unwrap()` on an `Err` value: External(ProtocolError { kind: Unknown, message: "missing required field PageHeader.type_" })

thread '' panicked at $HOME/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/parquet-55.1.0/src/encodings/rle.rs:485:58:
index out of bounds: the len is 50 but the index is 58
```

Contributor guide

Open the contributing guide

Research direction

Start at SerializedFileReader::get_row_group and RowGroupReader, then run the provided multithreaded example against the TPC-H Parquet file to reproduce the protocol errors and panic. Trace how shared std::fs::File handles are used while ColumnReader reads records. Done means determining whether this access pattern is supported and fixing or clearly documenting the correct usage.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.