parquet ArrowColumnWriter writer produces incorrect statistics for Decimal represented as ByteArray

[exyi]

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:

...
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.

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:

[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"] }