Multithreaded Reading Parquet from `std::fs::File`
- 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
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