parquet ArrowColumnWriter writer produces incorrect statistics for Decimal represented as ByteArray
- Dominant language
- Rust
- Stars
- 3.6k
- Forks
- 1.3k
- Avg merge
- 2d 14h
- Merged PRs (30d)
- 167
Description
### Describe the bug
I'm trying to write Parquet by converting from Arrow, and need to support arbitrary precision Decimal column backed by BYTE_ARRAY. I create arrow ByteArray from the encoded decimals and then use the ArrowColumnWriter machinery to write it to a Parquet file:
```rust
...
let array: ArrayPtr = Arc::new(BinaryArray::from_vec([...]));
let factory = ArrowRowGroupWriterFactory::new(&file, schema);
let mut writer: ArrowColumnWriter = factory.create_column_writers(0)?.remove(0);
let leaf = compute_leaves(&field, &array)?.remove(0);
writer.write(&leaf)?;
...
```
The column min/max statistics in the written Parquet files are wrong, probably because unsigned comparison is used (but Decimals are represented as two’s complement).
It only affects the Arrow path. When the `row_group.next_column().typed().write_batch(...)` API is used (as in #10860), it works correctly. It also only affects ByteArray, even FixedSizeBinaryArray works correctly.
### To Reproduce
The following code produces same Parquet file with a single decimal column. For simplicity, it's a 1 byte decimal with values -1, 0, -1.
```rust
use std::sync::Arc;
use arrow_array::{ArrayRef, BinaryArray, FixedSizeBinaryArray};
use arrow_schema::{Field, Schema};
use bytes::Bytes;
use parquet::arrow::arrow_writer::{ArrowColumnWriter, ArrowRowGroupWriterFactory, compute_leaves};
use parquet::basic::{LogicalType, Repetition, Type as PhysicalType};
use parquet::data_type::{ByteArray, ByteArrayType};
use parquet::errors::Result;
use parquet::file::reader::{FileReader, SerializedFileReader};
use parquet::file::writer::SerializedFileWriter;
use parquet::schema::types::Type;
// 1 byte decimal: -1, 0, 1
const VALUES: [&[u8]; 3] = [&[0xff], &[0x00], &[0x01]];
fn new_file(physical: PhysicalType) -> Result>> {
let mut field = Type::primitive_type_builder("value", physical)
.with_repetition(Repetition::REQUIRED)
.with_logical_type(Some(LogicalType::decimal(0, 2)))
.with_precision(2)
.with_scale(0);
if physical == PhysicalType::FIXED_LEN_BYTE_ARRAY {
field = field.with_length(1);
}
let root = Type::group_type_builder("root")
.with_fields(vec![Arc::new(field.build()?)])
.build()?;
SerializedFileWriter::new(Vec::new(), Arc::new(root), Default::default())
}
fn read_minmax(file: Vec) -> Result<(i8, i8)> {
let reader = SerializedFileReader::new(Bytes::from(file))?;
let metadata = reader.metadata();
assert_eq!(metadata.file_metadata().num_rows(), 3);
let stats = metadata.row_group(0).column(0).statistics().expect("missing statistics");
// every value is one byte
Ok((stats.min_bytes_opt().unwrap()[0] as i8,
stats.max_bytes_opt().unwrap()[0] as i8))
}
fn direct_parquet() -> Result> {
let mut file = new_file(PhysicalType::BYTE_ARRAY)?;
let mut group = file.next_row_group()?;
let mut column = group.next_column()?.unwrap();
let values = VALUES.iter().map(|v| ByteArray::from(v.to_vec())).collect::>();
column.typed::()
.write_batch(&values, None, None)?;
column.close()?;
group.close()?;
file.into_inner()
// finish_and_read_bounds(file)
}
fn arrow_binary() -> Result> {
let file = new_file(PhysicalType::BYTE_ARRAY)?;
let array = Arc::new(BinaryArray::from_vec(VALUES.to_vec()));
arrow_bounds(file, array)
}
fn arrow_fixed_size_binary() -> Result> {
let file = new_file(PhysicalType::FIXED_LEN_BYTE_ARRAY)?;
let array = Arc::new(FixedSizeBinaryArray::try_from_iter(VALUES.into_iter())?);
arrow_bounds(file, array)
}
fn arrow_bounds(mut file: SerializedFileWriter>, array: ArrayRef) -> Result> {
let field = Field::new("value", array.data_type().clone(), false);
let schema = Arc::new(Schema::new(vec![field.clone()]));
let factory = ArrowRowGroupWriterFactory::new(&file, schema);
let mut writer: ArrowColumnWriter = factory.create_column_writers(0)?.remove(0);
let leaf = compute_leaves(&field, &array)?.remove(0);
writer.write(&leaf)?;
let mut group = file.next_row_group()?;
writer.close()?.append_to_row_group(&mut group)?;
group.close()?;
file.into_inner()
}
fn main() {
let direct = read_minmax(direct_parquet().unwrap()).unwrap();
let fixed = read_minmax(arrow_fixed_size_binary().unwrap()).unwrap();
let binary = read_minmax(arrow_binary().unwrap()).unwrap();
println!("Direct Parquet BYTE_ARRAY: {direct:?}");
println!("Arrow FixedSizeBinary: {fixed:?}");
println!("Arrow Binary: {binary:?}");
assert_eq!(direct, (-1, 1), "incorrect direct Parquet");
assert_eq!(fixed, (-1, 1), "incorrect Arrow FixedSizeBinaryArray");
assert_eq!(binary, (-1, 1), "incorrect Arrow ByteArray");
}
```
### Expected behavior
All 3 approaches should get equal statistics: min: -1, max: 1
### Additional context
Tested on version 59.3. Cargo.toml:
```toml
[package]
name = "parquet-binary-decimal-statistics-repro"
version = "0.1.0"
edition = "2021"
publish = false
[dependencies]
arrow-array = "59.3"
arrow-schema = "59.3"
bytes = "1"
parquet = { version = "59.3", default-features = false, features = ["arrow"] }
```
Contributor guide
Research direction
Start with the ArrowColumnWriter.write path and its handling of BinaryArray statistics, then compare it with the direct ByteArrayType.write_batch path and the working FixedSizeBinaryArray case shown in the reproduction. Add a regression test using the supplied -1, 0, 1 decimal values; done means Arrow BYTE_ARRAY statistics report min -1 and max 1.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- databases
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 74/100