parquet: an ArrowReaderOptions::with_schema override from Binary to a string type skips UTF-8 validation #11140

// parquet/src/arrow/array_reader/byte_array.rs:192
let validate_utf8 = desc.converted_type() == ConvertedType::UTF8;

use std::sync::Arc;

use arrow::array::{Array, ArrayRef, BinaryArray, RecordBatch};
use arrow::datatypes::{DataType, Field, Schema};
use bytes::Bytes;
use parquet::arrow::arrow_reader::{ArrowReaderOptions, ParquetRecordBatchReaderBuilder};
use parquet::arrow::arrow_writer::{ArrowWriter, ArrowWriterOptions};

fn write() -> Bytes {
    // One BYTE_ARRAY column, no UTF8 annotation, one value of 0xff.
    let schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Binary, true)]));
    let array = Arc::new(BinaryArray::from(vec![Some(&[0xffu8][..])])) as ArrayRef;
    let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array]).unwrap();
    let mut buf = Vec::new();
    // Skip the advisory ARROW:schema key. The result is the same with it.
    let options = ArrowWriterOptions::new().with_skip_arrow_metadata(true);
    let mut writer = ArrowWriter::try_new_with_options(&mut buf, schema, options).unwrap();
    writer.write(&batch).unwrap();
    writer.close().unwrap();
    Bytes::from(buf)
}

fn read_as(data: Bytes, target: DataType) {
    let schema = Arc::new(Schema::new(vec![Field::new("b", target.clone(), true)]));
    let options = ArrowReaderOptions::new().with_schema(schema);
    let mut reader = ParquetRecordBatchReaderBuilder::try_new_with_options(data, options)
        .unwrap()
        .build()
        .unwrap();
    let batch = reader.next().unwrap().unwrap();
    let column = batch.column(0);
    println!("  read Ok, data_type = {}", column.data_type());
    match column.to_data().validate_full() {
        Ok(()) => println!("  validate_full: Ok"),
        Err(e) => println!("  validate_full: Err({e})"),
    }
}

fn main() {
    let data = write();
    for target in [DataType::Utf8, DataType::LargeUtf8, DataType::Utf8View] {
        println!("target = {target:?}");
        read_as(data.clone(), target);
    }
}