exyi opened a new issue, #11073:
URL: https://github.com/apache/arrow-rs/issues/11073
### 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<SerializedFileWriter<Vec<u8>>>
{
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<u8>) -> 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<Vec<u8>> {
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::<Vec<_>>();
column.typed::<ByteArrayType>()
.write_batch(&values, None, None)?;
column.close()?;
group.close()?;
file.into_inner()
// finish_and_read_bounds(file)
}
fn arrow_binary() -> Result<Vec<u8>> {
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<Vec<u8>> {
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<Vec<u8>>, array: ArrayRef) ->
Result<Vec<u8>> {
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"]
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
