apache / apache/arrow-rs

parquet ArrowColumnWriter writer produces incorrect statistics for Decimal represented as ByteArray

Open
#11,073 0 comments 0 reactions 0 assignees View on GitHub
bug
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.