[Rust] [Parquet] ParquetFileArrowReader fails to decode all pages if batches fall exactly on row group boundaries

[asfimport]

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

 ParquetError("Parquet error: Not all children array length are the same!")

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:

+-----+
| RG1 |
|     |
+-----+  <-- If a batch ends exactly at the end of this row group (page), RG2 is never read
+-----+
| RG2 |
|     |
+-----+

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:

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

The expected output is as follows (should always read 20 rows, regardless of the batch size):

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

Workaround

Use a different batch size that will not fall on record batch boundaries

Reporter: Andrew Lamb / @alamb
Assignee: Andrew Lamb / @alamb

Original Issue Attachments:

• parquet_file_arrow_reader.zip

PRs and other links:

• GitHub Pull Request #8007
• GitHub Pull Request #8009

_{Note: This issue was originally created as ARROW-9790. Please see the migration documentation for further details.}

[asfimport]

Andrew Lamb / @alamb:
I am pretty sure I know what the problem is, and I will have a PR up to fix this tomorrow morning

[asfimport]

Andrew Lamb / @alamb:
Here is the reproducer code, to save you from having to download it:

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
}

[asfimport]

Chao Sun / @sunchao:
Issue resolved by pull request 8007
#8007

[asfimport]

Andrew Lamb / @alamb:
See also apache/arrow-rs#2025