[
https://issues.apache.org/jira/browse/ARROW-9790?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17180510#comment-17180510
]
Andrew Lamb commented on ARROW-9790:
------------------------------------
Here is the reproducer code, to save you from having to download it:
{code}
use std::{fs, path::Path, rc::Rc};
use parquet::{
file::{
properties::WriterProperties,
writer::{FileWriter, SerializedFileWriter},
},
schema::parser::parse_message_type, column::writer::ColumnWriter,
};
use parquet::arrow::arrow_reader::ArrowReader;
use arrow::record_batch::RecordBatchReader;
fn main() {
make_repro_file();
println!("Size when reading with batch_size 100 : {}",
read_record_batches(100));
println!("Size when reading with batch_size 7 : {}",
read_record_batches(7));
println!("Size when reading with batch_size 5 : {}",
read_record_batches(5));
}
/// create repro.parquet file
fn make_repro_file() {
let path = Path::new("/tmp/repro.parquet");
let message_type = "
message schema {
REQUIRED INT32 b;
}
";
let schema = Rc::new(parse_message_type(message_type).unwrap());
let props = Rc::new(WriterProperties::builder().build());
let file = fs::File::create(&path).unwrap();
let mut writer = SerializedFileWriter::new(file, schema, props).unwrap();
let data = [1,2,3,4,5];
for _ in 0..4 {
let mut row_group_writer = writer.next_row_group().unwrap();
if let Some(mut col_writer) = row_group_writer.next_column().unwrap() {
match &mut col_writer {
ColumnWriter::Int32ColumnWriter(w) => {
w.write_batch(&data, None, None).expect("Writing data");
},
_=> unimplemented!("unexpected type")
}
row_group_writer.close_column(col_writer).unwrap();
}
writer.close_row_group(row_group_writer).unwrap();
}
writer.close().unwrap();
println!("wrote 20 rows in 4 row groups to /tmp/repro.parquet");
}
// returns the number of total rows, using batch_size
fn read_record_batches(batch_size: usize) -> usize {
//println!("Opening the file....");
let r = fs::File::open("/tmp/repro.parquet").unwrap();
let parquet_reader =
parquet::file::reader::SerializedFileReader::new(r).unwrap();
let mut reader =
parquet::arrow::arrow_reader::ParquetFileArrowReader::new(Rc::new(parquet_reader));
let mut record_batch_reader = reader.get_record_reader(batch_size).unwrap();
let mut total_rows = 0;
loop {
let rb = record_batch_reader.next_batch();
match rb {
Err(e) => println!("WARNING: error reading batch: {:?}, SKIPPING",
e),
Ok(Some(rb)) => {
//println!("Successfully got a new record batch of {} rows",
rb.num_rows());
total_rows += rb.num_rows();
},
Ok(None) => {
//println!("No more batches");
break;
}
}
}
total_rows
}
{code}
> [Rust] [Parquet] ParquetFileArrowReader fails to decode all pages if batches
> fall exactly on row group boundaries
> -----------------------------------------------------------------------------------------------------------------
>
> Key: ARROW-9790
> URL: https://issues.apache.org/jira/browse/ARROW-9790
> Project: Apache Arrow
> Issue Type: Bug
> Reporter: Andrew Lamb
> Assignee: Andrew Lamb
> Priority: Major
> Attachments: parquet_file_arrow_reader.zip
>
>
> When I was reading a parquet file into RecordBatches using
> {{ParquetFileArrowReader}} that had row groups that were 100,000 rows in
> length with a batch size of 60,000, after reading 300,000 rows successfully,
> I started seeing this error
> {code}
> ParquetError("Parquet error: Not all children array length are the same!")
> {code}
> Upon investigation, I found that when reading with
> {{ParquetFileArrowReader}}, if the parquet input file has multiple row
> groups, and if a batch happens to end at the end of a row group for Int or
> Float, no subsequent row groups are read
> Visually:
> {code}
> +-----+
> | RG1 |
> | |
> +-----+ <-- If a batch ends exactly at the end of this row group (page), RG2
> is never read
> +-----+
> | RG2 |
> | |
> +-----+
> {code}
> A reproducer is attached. 20 values should be read by the
> {{ParquetFileArrowReader}} regardless of the batch size. However, when using
> batch sizes such as {{5}} or {{3}} (which fall on a boundary between row
> groups) not all the rows are read.
> To run the reproducer, decompress the attachment
> [^parquet_file_arrow_reader.zip] and do `cargo run`
> The output is as follows:
> {code}
> wrote 20 rows in 4 row groups to /tmp/repro.parquet
> Size when reading with batch_size 100 : 20
> Size when reading with batch_size 7 : 20
> Size when reading with batch_size 5 : 5
> {code}
> The expected output is as follows (should always read 20 rows, regardless of
> the batch size):
> {code}
> wrote 20 rows in 4 row groups to /tmp/repro.parquet
> Size when reading with batch_size 100 : 20
> Size when reading with batch_size 7 : 20
> Size when reading with batch_size 5 : 20
> {code}
> h2. Workaround
> Use a different batch size that will not fall on record batch boundaries
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
