mz_arrow_util/

reader.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

//! Reader for [`arrow`] data that outputs [`Row`]s.

use std::sync::Arc;

use anyhow::Context;
use arrow::array::{
    Array, BinaryArray, BinaryViewArray, BooleanArray, Date32Array, Date64Array, Decimal128Array,
    Decimal256Array, FixedSizeBinaryArray, Float16Array, Float32Array, Float64Array, Int8Array,
    Int16Array, Int32Array, Int64Array, IntervalDayTimeArray, IntervalMonthDayNanoArray,
    IntervalYearMonthArray, LargeBinaryArray, LargeListArray, LargeStringArray, ListArray,
    MapArray, StringArray, StringViewArray, StructArray, Time32MillisecondArray, Time32SecondArray,
    Time64MicrosecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
    TimestampNanosecondArray, TimestampSecondArray, UInt8Array, UInt16Array, UInt32Array,
    UInt64Array,
};
use arrow::buffer::{NullBuffer, OffsetBuffer};
use arrow::datatypes::{DataType, IntervalUnit, TimeUnit};
use chrono::{DateTime, NaiveTime};
use dec::OrderedDecimal;
use itertools::Itertools;
use mz_ore::cast::CastFrom;
use mz_repr::adt::array::ArrayDimension;
use mz_repr::adt::date::Date;
use mz_repr::adt::interval::Interval;
use mz_repr::adt::jsonb::JsonbPacker;
use mz_repr::adt::numeric::{Numeric, rescale};
use mz_repr::adt::range::{Range, RangeLowerBound, RangeUpperBound};
use mz_repr::adt::timestamp::CheckedTimestamp;
use mz_repr::{Datum, RelationDesc, Row, RowPacker, SharedRow, SqlScalarType};
use ordered_float::OrderedFloat;
use uuid::Uuid;

use crate::mask_nulls;

/// Type that can read out of an [`arrow::array::StructArray`] and into a [`Row`], given a
/// [`RelationDesc`].
///
/// The inverse of a [`crate::builder::ArrowBuilder`].
///
/// Note: When creating an [`ArrowReader`] we perform a "one-time downcast" of the children Arrays
/// from the [`StructArray`], into `enum ColReader`s. This is a much more verbose approach than the
/// alternative of downcasting from a `dyn arrow::array::Array` every time we read a [`Row`], but
/// it is _much_ more performant.
pub struct ArrowReader {
    len: usize,
    readers: Vec<ColReader>,
}

impl ArrowReader {
    /// Create an [`ArrowReader`] validating that the provided [`RelationDesc`] and [`StructArray`]
    /// have a matching schema.
    ///
    /// The [`RelationDesc`] and [`StructArray`] need to uphold the following to be a valid pair:
    ///
    /// * Same number of columns.
    /// * Columns of all the same name.
    /// * Columns of compatible types.
    ///
    /// TODO(cf2): Relax some of these restrictions by allowing users to map column names, omit
    /// columns, perform some lightweight casting, and matching not on column name but column
    /// position.
    /// TODO(cf2): Allow specifying an optional `arrow::Schema` for extra metadata.
    pub fn new(desc: &RelationDesc, array: StructArray) -> Result<Self, anyhow::Error> {
        let inner_columns = array.columns();
        let desc_columns = desc.typ().columns();

        if inner_columns.len() != desc_columns.len() {
            return Err(anyhow::anyhow!(
                "wrong number of columns {} vs {}",
                inner_columns.len(),
                desc_columns.len()
            ));
        }

        let mut readers = Vec::with_capacity(desc_columns.len());
        for (col_name, col_type) in desc.iter() {
            let column = array
                .column_by_name(col_name)
                .ok_or_else(|| anyhow::anyhow!("'{col_name}' not found"))?;
            let reader = scalar_type_and_array_to_reader(&col_type.scalar_type, Arc::clone(column))
                .context(col_name.clone())?;

            readers.push(reader);
        }

        Ok(ArrowReader {
            len: array.len(),
            readers,
        })
    }

    /// Read the value at `idx` into the provided `Row`.
    pub fn read(&self, idx: usize, row: &mut Row) -> Result<(), anyhow::Error> {
        let mut packer = row.packer();
        for reader in &self.readers {
            reader.read(idx, &mut packer).context(idx)?;
        }
        Ok(())
    }

    /// Read all of the values in this [`ArrowReader`] into `rows`.
    pub fn read_all(&self, rows: &mut Vec<Row>) -> Result<usize, anyhow::Error> {
        for idx in 0..self.len {
            let mut row = Row::default();
            self.read(idx, &mut row).context(idx)?;
            rows.push(row);
        }
        Ok(self.len)
    }
}

fn scalar_type_and_array_to_reader(
    scalar_type: &SqlScalarType,
    array: Arc<dyn Array>,
) -> Result<ColReader, anyhow::Error> {
    fn downcast_array<T: arrow::array::Array + Clone + 'static>(array: Arc<dyn Array>) -> T {
        array
            .as_any()
            .downcast_ref::<T>()
            .expect("checked DataType")
            .clone()
    }

    match (scalar_type, array.data_type()) {
        (SqlScalarType::Bool, DataType::Boolean) => {
            Ok(ColReader::Boolean(downcast_array::<BooleanArray>(array)))
        }
        (SqlScalarType::Int16 | SqlScalarType::Int32 | SqlScalarType::Int64, DataType::Int8) => {
            let array = downcast_array::<Int8Array>(array);
            let cast: fn(i8) -> Datum<'static> = match scalar_type {
                SqlScalarType::Int16 => |x| Datum::Int16(i16::cast_from(x)),
                SqlScalarType::Int32 => |x| Datum::Int32(i32::cast_from(x)),
                SqlScalarType::Int64 => |x| Datum::Int64(i64::cast_from(x)),
                _ => unreachable!("checked above"),
            };
            Ok(ColReader::Int8 { array, cast })
        }
        (SqlScalarType::Int16, DataType::Int16) => {
            Ok(ColReader::Int16(downcast_array::<Int16Array>(array)))
        }
        (SqlScalarType::Int32, DataType::Int32) => {
            Ok(ColReader::Int32(downcast_array::<Int32Array>(array)))
        }
        (SqlScalarType::Int64, DataType::Int64) => {
            Ok(ColReader::Int64(downcast_array::<Int64Array>(array)))
        }
        (
            SqlScalarType::UInt16 | SqlScalarType::UInt32 | SqlScalarType::UInt64,
            DataType::UInt8,
        ) => {
            let array = downcast_array::<UInt8Array>(array);
            let cast: fn(u8) -> Datum<'static> = match scalar_type {
                SqlScalarType::UInt16 => |x| Datum::UInt16(u16::cast_from(x)),
                SqlScalarType::UInt32 => |x| Datum::UInt32(u32::cast_from(x)),
                SqlScalarType::UInt64 => |x| Datum::UInt64(u64::cast_from(x)),
                _ => unreachable!("checked above"),
            };
            Ok(ColReader::UInt8 { array, cast })
        }
        (SqlScalarType::UInt16, DataType::UInt16) => {
            Ok(ColReader::UInt16(downcast_array::<UInt16Array>(array)))
        }
        // `oid` shares `Datum::UInt32`'s representation, so it decodes identically.
        (SqlScalarType::UInt32 | SqlScalarType::Oid, DataType::UInt32) => {
            Ok(ColReader::UInt32(downcast_array::<UInt32Array>(array)))
        }
        (SqlScalarType::UInt64, DataType::UInt64) => {
            Ok(ColReader::UInt64(downcast_array::<UInt64Array>(array)))
        }
        (SqlScalarType::MzTimestamp, DataType::UInt64) => {
            Ok(ColReader::MzTimestamp(downcast_array::<UInt64Array>(array)))
        }
        (SqlScalarType::Float32 | SqlScalarType::Float64, DataType::Float16) => {
            let array = downcast_array::<Float16Array>(array);
            let cast: fn(half::f16) -> Datum<'static> = match scalar_type {
                SqlScalarType::Float32 => |x| Datum::Float32(OrderedFloat::from(x.to_f32())),
                SqlScalarType::Float64 => |x| Datum::Float64(OrderedFloat::from(x.to_f64())),
                _ => unreachable!("checked above"),
            };
            Ok(ColReader::Float16 { array, cast })
        }
        (SqlScalarType::Float32, DataType::Float32) => {
            Ok(ColReader::Float32(downcast_array::<Float32Array>(array)))
        }
        (SqlScalarType::Float64, DataType::Float64) => {
            Ok(ColReader::Float64(downcast_array::<Float64Array>(array)))
        }
        (SqlScalarType::Numeric { max_scale }, DataType::Decimal128(precision, scale)) => {
            use num_traits::Pow;

            let base = Numeric::from(10);
            let scale = Numeric::from(*scale);
            let scale_factor = base.pow(scale);

            let precision = usize::cast_from(*precision);
            // Don't use the context here, but make sure the precision is valid.
            let mut ctx = dec::Context::<Numeric>::default();
            ctx.set_precision(precision).map_err(|e| {
                anyhow::anyhow!("invalid precision from Decimal128, {precision}, {e}")
            })?;

            let array = downcast_array::<Decimal128Array>(array);

            Ok(ColReader::Decimal128 {
                array,
                scale_factor,
                precision,
                destination_max_scale: (*max_scale).map(|s| s.into_u8()),
            })
        }
        (SqlScalarType::Numeric { max_scale }, DataType::Decimal256(precision, scale)) => {
            use num_traits::Pow;

            let base = Numeric::from(10);
            let scale = Numeric::from(*scale);
            let scale_factor = base.pow(scale);

            let precision = usize::cast_from(*precision);
            // Don't use the context here, but make sure the precision is valid.
            let mut ctx = dec::Context::<Numeric>::default();
            ctx.set_precision(precision).map_err(|e| {
                anyhow::anyhow!("invalid precision from Decimal256, {precision}, {e}")
            })?;

            let array = downcast_array::<Decimal256Array>(array);

            Ok(ColReader::Decimal256 {
                array,
                scale_factor,
                precision,
                destination_max_scale: (*max_scale).map(|s| s.into_u8()),
            })
        }
        (SqlScalarType::Bytes, DataType::Binary) => {
            Ok(ColReader::Binary(downcast_array::<BinaryArray>(array)))
        }
        (SqlScalarType::Bytes, DataType::LargeBinary) => {
            let array = downcast_array::<LargeBinaryArray>(array);
            Ok(ColReader::LargeBinary(array))
        }
        (SqlScalarType::Bytes, DataType::FixedSizeBinary(_)) => {
            let array = downcast_array::<FixedSizeBinaryArray>(array);
            Ok(ColReader::FixedSizeBinary(array))
        }
        (SqlScalarType::Bytes, DataType::BinaryView) => {
            let array = downcast_array::<BinaryViewArray>(array);
            Ok(ColReader::BinaryView(array))
        }
        (
            SqlScalarType::Uuid,
            DataType::Binary
            | DataType::BinaryView
            | DataType::LargeBinary
            | DataType::FixedSizeBinary(_),
        ) => {
            let reader = scalar_type_and_array_to_reader(&SqlScalarType::Bytes, array)
                .context("uuid reader")?;
            Ok(ColReader::Uuid(Box::new(reader)))
        }
        // `char` and `varchar` store `Datum::String`, so they decode as strings.
        // The writer emits them as Utf8/LargeUtf8 depending on length.
        (
            SqlScalarType::String | SqlScalarType::Char { .. } | SqlScalarType::VarChar { .. },
            DataType::Utf8,
        ) => Ok(ColReader::String(downcast_array::<StringArray>(array))),
        (
            SqlScalarType::String | SqlScalarType::Char { .. } | SqlScalarType::VarChar { .. },
            DataType::LargeUtf8,
        ) => {
            let array = downcast_array::<LargeStringArray>(array);
            Ok(ColReader::LargeString(array))
        }
        (
            SqlScalarType::String | SqlScalarType::Char { .. } | SqlScalarType::VarChar { .. },
            DataType::Utf8View,
        ) => {
            let array = downcast_array::<StringViewArray>(array);
            Ok(ColReader::StringView(array))
        }
        (SqlScalarType::Jsonb, DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View) => {
            let reader = scalar_type_and_array_to_reader(&SqlScalarType::String, array)
                .context("json reader")?;
            Ok(ColReader::Jsonb(Box::new(reader)))
        }
        (SqlScalarType::Timestamp { .. }, DataType::Timestamp(TimeUnit::Second, None)) => {
            let array = downcast_array::<TimestampSecondArray>(array);
            Ok(ColReader::TimestampSecond(array))
        }
        (SqlScalarType::Timestamp { .. }, DataType::Timestamp(TimeUnit::Millisecond, None)) => {
            let array = downcast_array::<TimestampMillisecondArray>(array);
            Ok(ColReader::TimestampMillisecond(array))
        }
        (SqlScalarType::Timestamp { .. }, DataType::Timestamp(TimeUnit::Microsecond, None)) => {
            let array = downcast_array::<TimestampMicrosecondArray>(array);
            Ok(ColReader::TimestampMicrosecond(array))
        }
        (SqlScalarType::Timestamp { .. }, DataType::Timestamp(TimeUnit::Nanosecond, None)) => {
            let array = downcast_array::<TimestampNanosecondArray>(array);
            Ok(ColReader::TimestampNanosecond(array))
        }
        // A tz-aware timestamp array stores UTC-normalized instants, so the tz
        // string is metadata we can ignore. The writer always emits microseconds.
        (
            SqlScalarType::TimestampTz { .. },
            DataType::Timestamp(TimeUnit::Microsecond, Some(_)),
        ) => {
            let array = downcast_array::<TimestampMicrosecondArray>(array);
            Ok(ColReader::TimestampTzMicrosecond(array))
        }
        (SqlScalarType::Date, DataType::Date32) => {
            let array = downcast_array::<Date32Array>(array);
            Ok(ColReader::Date32(array))
        }
        (SqlScalarType::Date, DataType::Date64) => {
            let array = downcast_array::<Date64Array>(array);
            Ok(ColReader::Date64(array))
        }
        (SqlScalarType::Time, DataType::Time32(TimeUnit::Second)) => {
            let array = downcast_array::<Time32SecondArray>(array);
            Ok(ColReader::Time32Seconds(array))
        }
        (SqlScalarType::Time, DataType::Time32(TimeUnit::Millisecond)) => {
            let array = downcast_array::<Time32MillisecondArray>(array);
            Ok(ColReader::Time32Milliseconds(array))
        }
        (SqlScalarType::Time, DataType::Time64(TimeUnit::Microsecond)) => {
            let array = downcast_array::<Time64MicrosecondArray>(array);
            Ok(ColReader::Time64Microseconds(array))
        }
        (
            SqlScalarType::List {
                element_type,
                custom_id: _,
            },
            DataType::List(_),
        ) => {
            let array = downcast_array::<ListArray>(array);
            let inner_decoder =
                scalar_type_and_array_to_reader(element_type, Arc::clone(array.values()))
                    .context("list")?;
            Ok(ColReader::List {
                offsets: array.offsets().clone(),
                values: Box::new(inner_decoder),
                nulls: array.nulls().cloned(),
            })
        }
        (
            SqlScalarType::List {
                element_type,
                custom_id: _,
            },
            DataType::LargeList(_),
        ) => {
            let array = downcast_array::<LargeListArray>(array);
            let inner_decoder =
                scalar_type_and_array_to_reader(element_type, Arc::clone(array.values()))
                    .context("large list")?;
            Ok(ColReader::LargeList {
                offsets: array.offsets().clone(),
                values: Box::new(inner_decoder),
                nulls: array.nulls().cloned(),
            })
        }
        (SqlScalarType::Array(element_type), DataType::Struct(_)) => {
            // The builder encodes an array as a struct of an `items` list (the
            // flat, row-major elements) and a `dimensions` count. Reverse that.
            let struct_array = downcast_array::<StructArray>(array);

            let items = struct_array
                .column_by_name("items")
                .ok_or_else(|| anyhow::anyhow!("array struct missing 'items' field"))?;
            let items = items
                .as_any()
                .downcast_ref::<ListArray>()
                .ok_or_else(|| anyhow::anyhow!("array 'items' field is not a List"))?;
            let values = scalar_type_and_array_to_reader(element_type, Arc::clone(items.values()))
                .context("array items")?;

            let dims_col = struct_array
                .column_by_name("dimensions")
                .ok_or_else(|| anyhow::anyhow!("array struct missing 'dimensions' field"))?;
            // The builder writes `dimensions` as UInt8. An Iceberg round-trip
            // widens it to Int32 (Iceberg has no narrow integer types), so
            // accept both.
            let dims = match dims_col.data_type() {
                DataType::UInt8 => {
                    ArrayDims::UInt8(downcast_array::<UInt8Array>(Arc::clone(dims_col)))
                }
                DataType::Int32 => {
                    ArrayDims::Int32(downcast_array::<Int32Array>(Arc::clone(dims_col)))
                }
                other => anyhow::bail!("unsupported array 'dimensions' type: {other:?}"),
            };

            Ok(ColReader::Array {
                offsets: items.offsets().clone(),
                values: Box::new(values),
                dims,
                nulls: struct_array.nulls().cloned(),
            })
        }
        (
            SqlScalarType::Record {
                fields,
                custom_id: _,
            },
            DataType::Struct(_),
        ) => {
            let record_array = downcast_array::<StructArray>(array);
            let null_mask = record_array.nulls();

            let mut decoders = Vec::with_capacity(fields.len());
            for (name, typ) in fields.iter() {
                let inner_array = record_array
                    .column_by_name(name)
                    .ok_or_else(|| anyhow::anyhow!("missing name '{name}'"))?;
                let inner_array = mask_nulls(inner_array, null_mask);
                let inner_decoder = scalar_type_and_array_to_reader(&typ.scalar_type, inner_array)
                    .context(name.clone())?;

                decoders.push(Box::new(inner_decoder));
            }

            Ok(ColReader::Record {
                fields: decoders,
                nulls: null_mask.cloned(),
            })
        }
        (
            SqlScalarType::Map {
                value_type,
                custom_id: _,
            },
            DataType::Map(_, _),
        ) => {
            let map_array = downcast_array::<MapArray>(array);
            let keys = map_array
                .keys()
                .as_any()
                .downcast_ref::<StringArray>()
                .expect("map keys should be Utf8 strings")
                .clone();
            let values_reader =
                scalar_type_and_array_to_reader(value_type, Arc::clone(map_array.values()))
                    .context("map values")?;
            Ok(ColReader::Map {
                offsets: map_array.offsets().clone(),
                keys,
                values: Box::new(values_reader),
                nulls: map_array.nulls().cloned(),
            })
        }
        (SqlScalarType::Range { element_type }, DataType::Struct(_)) => {
            let struct_array = downcast_array::<StructArray>(array);
            let lower_array = struct_array
                .column_by_name("lower")
                .ok_or_else(|| anyhow::anyhow!("range struct missing 'lower' field"))?;
            let upper_array = struct_array
                .column_by_name("upper")
                .ok_or_else(|| anyhow::anyhow!("range struct missing 'upper' field"))?;
            let empty_array = struct_array
                .column_by_name("empty")
                .ok_or_else(|| anyhow::anyhow!("range struct missing 'empty' field"))?;
            let lower_inclusive_array = struct_array
                .column_by_name("lower_inclusive")
                .ok_or_else(|| anyhow::anyhow!("range struct missing 'lower_inclusive' field"))?;
            let upper_inclusive_array = struct_array
                .column_by_name("upper_inclusive")
                .ok_or_else(|| anyhow::anyhow!("range struct missing 'upper_inclusive' field"))?;

            let lower_reader =
                scalar_type_and_array_to_reader(element_type, Arc::clone(lower_array))
                    .context("range lower")?;
            let upper_reader =
                scalar_type_and_array_to_reader(element_type, Arc::clone(upper_array))
                    .context("range upper")?;

            let empty = downcast_array::<BooleanArray>(Arc::clone(empty_array));
            let lower_inclusive = downcast_array::<BooleanArray>(Arc::clone(lower_inclusive_array));
            let upper_inclusive = downcast_array::<BooleanArray>(Arc::clone(upper_inclusive_array));

            let lower_nulls = lower_array.nulls().cloned();
            let upper_nulls = upper_array.nulls().cloned();

            Ok(ColReader::Range {
                lower: Box::new(lower_reader),
                lower_nulls,
                upper: Box::new(upper_reader),
                upper_nulls,
                lower_inclusive,
                upper_inclusive,
                empty,
                nulls: struct_array.nulls().cloned(),
            })
        }
        (SqlScalarType::Interval, DataType::Interval(IntervalUnit::YearMonth)) => {
            Ok(ColReader::IntervalYearMonth(downcast_array::<
                IntervalYearMonthArray,
            >(array)))
        }
        (SqlScalarType::Interval, DataType::Interval(IntervalUnit::DayTime)) => {
            Ok(ColReader::IntervalDayTime(downcast_array::<
                IntervalDayTimeArray,
            >(array)))
        }
        (SqlScalarType::Interval, DataType::Interval(IntervalUnit::MonthDayNano)) => {
            Ok(ColReader::IntervalMonthDayNano(downcast_array::<
                IntervalMonthDayNanoArray,
            >(array)))
        }
        other => anyhow::bail!("unsupported: {other:?}"),
    }
}

/// The `dimensions` field of an encoded array. The builder writes it as
/// [`UInt8`](DataType::UInt8); an Iceberg round-trip widens it to
/// [`Int32`](DataType::Int32).
enum ArrayDims {
    UInt8(UInt8Array),
    Int32(Int32Array),
}

impl ArrayDims {
    /// The number of dimensions of the array at `idx`.
    fn ndims(&self, idx: usize) -> Result<u8, anyhow::Error> {
        match self {
            ArrayDims::UInt8(array) => Ok(array.value(idx)),
            ArrayDims::Int32(array) => {
                u8::try_from(array.value(idx)).context("array dimension count out of range")
            }
        }
    }
}

/// A "downcasted" version of [`arrow::array::Array`] that supports reading [`Datum`]s.
///
/// Note: While this is fairly verbose, one-time "downcasting" to an enum is _much_ more performant
/// than downcasting every time we read a [`Datum`].
enum ColReader {
    Boolean(arrow::array::BooleanArray),

    Int8 {
        array: arrow::array::Int8Array,
        cast: fn(i8) -> Datum<'static>,
    },
    Int16(arrow::array::Int16Array),
    Int32(arrow::array::Int32Array),
    Int64(arrow::array::Int64Array),

    UInt8 {
        array: arrow::array::UInt8Array,
        cast: fn(u8) -> Datum<'static>,
    },
    UInt16(arrow::array::UInt16Array),
    UInt32(arrow::array::UInt32Array),
    UInt64(arrow::array::UInt64Array),

    Float16 {
        array: arrow::array::Float16Array,
        cast: fn(half::f16) -> Datum<'static>,
    },
    Float32(arrow::array::Float32Array),
    Float64(arrow::array::Float64Array),

    Decimal128 {
        array: Decimal128Array,
        scale_factor: Numeric,
        precision: usize,
        destination_max_scale: Option<u8>,
    },
    Decimal256 {
        array: Decimal256Array,
        scale_factor: Numeric,
        precision: usize,
        destination_max_scale: Option<u8>,
    },

    Binary(arrow::array::BinaryArray),
    LargeBinary(arrow::array::LargeBinaryArray),
    FixedSizeBinary(arrow::array::FixedSizeBinaryArray),
    BinaryView(arrow::array::BinaryViewArray),
    Uuid(Box<ColReader>),

    String(arrow::array::StringArray),
    LargeString(arrow::array::LargeStringArray),
    StringView(arrow::array::StringViewArray),
    Jsonb(Box<ColReader>),

    TimestampSecond(arrow::array::TimestampSecondArray),
    TimestampMillisecond(arrow::array::TimestampMillisecondArray),
    TimestampMicrosecond(arrow::array::TimestampMicrosecondArray),
    TimestampNanosecond(arrow::array::TimestampNanosecondArray),
    TimestampTzMicrosecond(arrow::array::TimestampMicrosecondArray),

    MzTimestamp(arrow::array::UInt64Array),

    Date32(Date32Array),
    Date64(Date64Array),

    Time32Seconds(Time32SecondArray),
    Time32Milliseconds(arrow::array::Time32MillisecondArray),
    Time64Microseconds(arrow::array::Time64MicrosecondArray),

    List {
        offsets: OffsetBuffer<i32>,
        values: Box<ColReader>,
        nulls: Option<NullBuffer>,
    },
    LargeList {
        offsets: OffsetBuffer<i64>,
        values: Box<ColReader>,
        nulls: Option<NullBuffer>,
    },

    Array {
        offsets: OffsetBuffer<i32>,
        values: Box<ColReader>,
        dims: ArrayDims,
        nulls: Option<NullBuffer>,
    },

    Record {
        fields: Vec<Box<ColReader>>,
        nulls: Option<NullBuffer>,
    },

    Map {
        offsets: OffsetBuffer<i32>,
        keys: StringArray,
        values: Box<ColReader>,
        nulls: Option<NullBuffer>,
    },

    Range {
        lower: Box<ColReader>,
        lower_nulls: Option<NullBuffer>,
        upper: Box<ColReader>,
        upper_nulls: Option<NullBuffer>,
        lower_inclusive: BooleanArray,
        upper_inclusive: BooleanArray,
        empty: BooleanArray,
        nulls: Option<NullBuffer>,
    },

    IntervalYearMonth(IntervalYearMonthArray),
    IntervalDayTime(IntervalDayTimeArray),
    IntervalMonthDayNano(IntervalMonthDayNanoArray),
}

impl ColReader {
    fn read(&self, idx: usize, packer: &mut RowPacker) -> Result<(), anyhow::Error> {
        let datum = match self {
            ColReader::Boolean(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|x| if x { Datum::True } else { Datum::False }),
            ColReader::Int8 { array, cast } => {
                array.is_valid(idx).then(|| array.value(idx)).map(cast)
            }
            ColReader::Int16(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::Int16),
            ColReader::Int32(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::Int32),
            ColReader::Int64(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::Int64),
            ColReader::UInt8 { array, cast } => {
                array.is_valid(idx).then(|| array.value(idx)).map(cast)
            }
            ColReader::UInt16(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::UInt16),
            ColReader::UInt32(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::UInt32),
            ColReader::UInt64(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::UInt64),
            ColReader::Float16 { array, cast } => {
                array.is_valid(idx).then(|| array.value(idx)).map(cast)
            }
            ColReader::Float32(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|x| Datum::Float32(OrderedFloat(x))),
            ColReader::Float64(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|x| Datum::Float64(OrderedFloat(x))),
            ColReader::Decimal128 {
                array,
                scale_factor,
                precision,
                destination_max_scale,
            } => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|x| {
                    // Create a Numeric from our i128 with precision.
                    let mut ctx = dec::Context::<Numeric>::default();
                    ctx.set_precision(*precision).expect("checked before");
                    let mut num = ctx.from_i128(x);

                    // Scale the number.
                    ctx.div(&mut num, scale_factor);

                    if let Some(destination_max_scale) = destination_max_scale {
                        rescale(&mut num, *destination_max_scale)?;
                    }

                    Ok::<_, anyhow::Error>(Datum::Numeric(OrderedDecimal(num)))
                })
                .transpose()?,
            ColReader::Decimal256 {
                array,
                scale_factor,
                precision,
                destination_max_scale,
            } => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|x| {
                    let s = x.to_string();

                    // Parse a i256 from it's String representation.
                    //
                    // TODO(cf3): See if we can add support for 256-bit numbers to the `dec` crate.
                    let mut ctx = dec::Context::<Numeric>::default();
                    ctx.set_precision(*precision).expect("checked before");
                    let mut num = ctx
                        .parse(s)
                        .map_err(|e| anyhow::anyhow!("decimal out of range: {e}"))?;

                    // Scale the number.
                    ctx.div(&mut num, scale_factor);

                    if let Some(destination_max_scale) = destination_max_scale {
                        rescale(&mut num, *destination_max_scale)?;
                    }

                    Ok::<_, anyhow::Error>(Datum::Numeric(OrderedDecimal(num)))
                })
                .transpose()?,
            ColReader::Binary(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::Bytes),
            ColReader::LargeBinary(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::Bytes),
            ColReader::FixedSizeBinary(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::Bytes),
            ColReader::BinaryView(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::Bytes),
            ColReader::Uuid(reader) => {
                // First read a binary value into a temp row, and later parse that as UUID into our
                // actual Row Packer.
                let mut temp_row = SharedRow::get();
                reader.read(idx, &mut temp_row.packer()).context("uuid")?;
                let slice = match temp_row.unpack_first() {
                    Datum::Bytes(slice) => slice,
                    Datum::Null => {
                        packer.push(Datum::Null);
                        return Ok(());
                    }
                    other => anyhow::bail!("expected String, found {other:?}"),
                };

                let uuid = Uuid::from_slice(slice).context("parsing uuid")?;
                Some(Datum::Uuid(uuid))
            }
            ColReader::String(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::String),
            ColReader::LargeString(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::String),
            ColReader::StringView(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(Datum::String),
            ColReader::Jsonb(reader) => {
                // First read a string value into a temp row, and later parse that as JSON into our
                // actual Row Packer.
                let mut temp_row = SharedRow::get();
                reader.read(idx, &mut temp_row.packer()).context("jsonb")?;
                let value = match temp_row.unpack_first() {
                    Datum::String(value) => value,
                    Datum::Null => {
                        packer.push(Datum::Null);
                        return Ok(());
                    }
                    other => anyhow::bail!("expected String, found {other:?}"),
                };

                JsonbPacker::new(packer)
                    .pack_str(value)
                    .context("roundtrip json")?;

                // Return early because we've already packed the necessasry Datums.
                return Ok(());
            }
            ColReader::TimestampSecond(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|secs| {
                    let dt = DateTime::from_timestamp(secs, 0)
                        .ok_or_else(|| anyhow::anyhow!("invalid timestamp seconds {secs}"))?;
                    let dt = CheckedTimestamp::from_timestamplike(dt.naive_utc())
                        .context("TimestampSeconds")?;
                    Ok::<_, anyhow::Error>(Datum::Timestamp(dt))
                })
                .transpose()?,
            ColReader::TimestampMillisecond(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|millis| {
                    let dt = DateTime::from_timestamp_millis(millis).ok_or_else(|| {
                        anyhow::anyhow!("invalid timestamp milliseconds {millis}")
                    })?;
                    let dt = CheckedTimestamp::from_timestamplike(dt.naive_utc())
                        .context("TimestampMillis")?;
                    Ok::<_, anyhow::Error>(Datum::Timestamp(dt))
                })
                .transpose()?,
            ColReader::TimestampMicrosecond(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|micros| {
                    let dt = DateTime::from_timestamp_micros(micros).ok_or_else(|| {
                        anyhow::anyhow!("invalid timestamp microseconds {micros}")
                    })?;
                    let dt = CheckedTimestamp::from_timestamplike(dt.naive_utc())
                        .context("TimestampMicros")?;
                    Ok::<_, anyhow::Error>(Datum::Timestamp(dt))
                })
                .transpose()?,
            ColReader::TimestampNanosecond(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|nanos| {
                    let dt = DateTime::from_timestamp_nanos(nanos);
                    let dt = CheckedTimestamp::from_timestamplike(dt.naive_utc())
                        .context("TimestampNanos")?;
                    Ok::<_, anyhow::Error>(Datum::Timestamp(dt))
                })
                .transpose()?,
            ColReader::TimestampTzMicrosecond(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|micros| {
                    let dt = DateTime::from_timestamp_micros(micros).ok_or_else(|| {
                        anyhow::anyhow!("invalid timestamptz microseconds {micros}")
                    })?;
                    let dt =
                        CheckedTimestamp::from_timestamplike(dt).context("TimestampTzMicros")?;
                    Ok::<_, anyhow::Error>(Datum::TimestampTz(dt))
                })
                .transpose()?,
            ColReader::MzTimestamp(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|v| Datum::MzTimestamp(mz_repr::Timestamp::from(v))),
            ColReader::Date32(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|unix_days| {
                    let date = Date::from_unix_epoch(unix_days).context("date32")?;
                    Ok::<_, anyhow::Error>(Datum::Date(date))
                })
                .transpose()?,
            ColReader::Date64(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|unix_millis| {
                    let date = DateTime::from_timestamp_millis(unix_millis)
                        .ok_or_else(|| anyhow::anyhow!("invalid Date64 {unix_millis}"))?;
                    let unix_epoch = DateTime::from_timestamp(0, 0)
                        .expect("UNIX epoch")
                        .date_naive();
                    let delta = date.date_naive().signed_duration_since(unix_epoch);
                    let days: i32 = delta.num_days().try_into().context("date64")?;
                    let date = Date::from_unix_epoch(days).context("date64")?;
                    Ok::<_, anyhow::Error>(Datum::Date(date))
                })
                .transpose()?,
            ColReader::Time32Seconds(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|secs| {
                    let usecs: u32 = secs.try_into().context("time32 seconds")?;
                    let time = NaiveTime::from_num_seconds_from_midnight_opt(usecs, 0)
                        .ok_or_else(|| anyhow::anyhow!("invalid Time32 Seconds {secs}"))?;
                    Ok::<_, anyhow::Error>(Datum::Time(time))
                })
                .transpose()?,
            ColReader::Time32Milliseconds(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|millis| {
                    let umillis: u32 = millis.try_into().context("time32 milliseconds")?;
                    let usecs = umillis / 1000;
                    let unanos = (umillis % 1000).saturating_mul(1_000_000);
                    let time = NaiveTime::from_num_seconds_from_midnight_opt(usecs, unanos)
                        .ok_or_else(|| anyhow::anyhow!("invalid Time32 Milliseconds {umillis}"))?;
                    Ok::<_, anyhow::Error>(Datum::Time(time))
                })
                .transpose()?,
            ColReader::Time64Microseconds(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|micros| {
                    // Inverse of the builder's `secs * 1_000_000 + nanos / 1_000`.
                    let secs: u32 = (micros / 1_000_000).try_into().context("time64 seconds")?;
                    let nanos: u32 = (micros % 1_000_000)
                        .try_into()
                        .map(|us: u32| us.saturating_mul(1_000))
                        .context("time64 microseconds")?;
                    let time = NaiveTime::from_num_seconds_from_midnight_opt(secs, nanos)
                        .ok_or_else(|| anyhow::anyhow!("invalid Time64 Microseconds {micros}"))?;
                    Ok::<_, anyhow::Error>(Datum::Time(time))
                })
                .transpose()?,
            ColReader::List {
                offsets,
                values,
                nulls,
            } => {
                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
                if !is_valid {
                    packer.push(Datum::Null);
                    return Ok(());
                }

                let start: usize = offsets[idx].try_into().context("list start offset")?;
                let end: usize = offsets[idx + 1].try_into().context("list end offset")?;

                packer
                    .push_list_with(|packer| {
                        for idx in start..end {
                            values.read(idx, packer)?;
                        }
                        Ok::<_, anyhow::Error>(())
                    })
                    .context("pack list")?;

                // Return early because we've already packed the necessasry Datums.
                return Ok(());
            }
            ColReader::LargeList {
                offsets,
                values,
                nulls,
            } => {
                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
                if !is_valid {
                    packer.push(Datum::Null);
                    return Ok(());
                }

                let start: usize = offsets[idx].try_into().context("list start offset")?;
                let end: usize = offsets[idx + 1].try_into().context("list end offset")?;

                packer
                    .push_list_with(|packer| {
                        for idx in start..end {
                            values.read(idx, packer)?;
                        }
                        Ok::<_, anyhow::Error>(())
                    })
                    .context("pack list")?;

                // Return early because we've already packed the necessasry Datums.
                return Ok(());
            }
            ColReader::Array {
                offsets,
                values,
                dims,
                nulls,
            } => {
                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
                if !is_valid {
                    packer.push(Datum::Null);
                    return Ok(());
                }

                let start: usize = offsets[idx].try_into().context("array start offset")?;
                let end: usize = offsets[idx + 1].try_into().context("array end offset")?;
                let nelements = end - start;

                // The encoding stores only the dimension count and the flat,
                // row-major elements (see the builder), so per-dimension extents
                // are recoverable only for 0- and 1-dimensional arrays. A
                // higher-dimensional array's extents cannot be reconstructed, so
                // reject it rather than guess a shape.
                let ndims = dims.ndims(idx)?;
                if ndims > 1 {
                    anyhow::bail!(
                        "cannot decode {ndims}-dimensional array from parquet: the encoding \
                         records only the dimension count, not per-dimension extents"
                    );
                }
                let one_dim = [ArrayDimension {
                    lower_bound: 1,
                    length: nelements,
                }];
                let array_dims: &[ArrayDimension] = if ndims == 0 { &[] } else { &one_dim };

                // SAFETY: the closure returns exactly the number of elements it
                // pushes (`end - start`).
                unsafe {
                    packer.push_array_with_unchecked(array_dims, |packer| {
                        for idx in start..end {
                            values.read(idx, packer)?;
                        }
                        Ok::<_, anyhow::Error>(end - start)
                    })
                }
                .context("pack array")?;

                // Return early because we've already packed the necessasry Datums.
                return Ok(());
            }
            ColReader::Record { fields, nulls } => {
                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
                if !is_valid {
                    packer.push(Datum::Null);
                    return Ok(());
                }

                packer
                    .push_list_with(|packer| {
                        for field in fields {
                            field.read(idx, packer)?;
                        }
                        Ok::<_, anyhow::Error>(())
                    })
                    .context("pack record")?;

                // Return early because we've already packed the necessasry Datums.
                return Ok(());
            }
            ColReader::Map {
                offsets,
                keys,
                values,
                nulls,
            } => {
                let is_non_null = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
                if !is_non_null {
                    packer.push(Datum::Null);
                    return Ok(());
                }

                let start: usize = offsets[idx].try_into().context("map start offset")?;
                let end: usize = offsets[idx + 1].try_into().context("map end offset")?;

                // Arrow's MapArray doesn't guarantee that keys are in sorted order, but Materialize's
                // Datum::Map does, so we need to sort the keys here before packing them, or else
                // many assumptions will break.
                let mut kv_sorted = (start..end)
                    .map(|i| (keys.value(i), i))
                    .sorted_by_key(|(k, _)| *k)
                    .peekable();

                packer
                    .push_dict_with(|packer| {
                        while let Some((key, i)) = kv_sorted.next() {
                            // Parquet docs state that if there are duplicate keys, the last value
                            // should be used, so skip duplicates here.
                            //
                            // sorted_by_key is a stable sort, so entries with duplicate keys will
                            // maintain their original order, and we can pick the last one here.
                            if let Some((next_key, _)) = kv_sorted.peek() {
                                if key == *next_key {
                                    continue;
                                }
                            }
                            packer.push(Datum::String(key));
                            values.read(i, packer)?;
                        }
                        Ok::<_, anyhow::Error>(())
                    })
                    .context("pack map")?;

                // Return early because we've already packed the necessary Datums.
                return Ok(());
            }
            ColReader::Range {
                lower,
                lower_nulls,
                upper,
                upper_nulls,
                lower_inclusive,
                upper_inclusive,
                empty,
                nulls,
            } => {
                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
                if !is_valid {
                    packer.push(Datum::Null);
                    return Ok(());
                }

                if empty.value(idx) {
                    packer.push(Datum::Range(Range { inner: None }));
                    return Ok(());
                }

                let lower_is_infinite = lower_nulls
                    .as_ref()
                    .map(|n| !n.is_valid(idx))
                    .unwrap_or(false);
                let upper_is_infinite = upper_nulls
                    .as_ref()
                    .map(|n| !n.is_valid(idx))
                    .unwrap_or(false);

                // Read finite bounds into owned Rows that live for the rest of
                // this block, so the Datums we borrow out of them stay valid
                // for the `push_range` call below.
                let lower_row = if lower_is_infinite {
                    None
                } else {
                    let mut temp = SharedRow::get();
                    lower.read(idx, &mut temp.packer())?;
                    Some(temp.clone())
                };
                let upper_row = if upper_is_infinite {
                    None
                } else {
                    let mut temp = SharedRow::get();
                    upper.read(idx, &mut temp.packer())?;
                    Some(temp.clone())
                };

                let lower_bound = RangeLowerBound {
                    inclusive: lower_inclusive.value(idx),
                    bound: lower_row.as_ref().map(|row| row.unpack_first()),
                };
                let upper_bound = RangeUpperBound {
                    inclusive: upper_inclusive.value(idx),
                    bound: upper_row.as_ref().map(|row| row.unpack_first()),
                };

                // Use `push_range` (not `push_range_with`) so the range is
                // canonicalized before being packed. Parquet files authored by
                // external engines may encode discrete ranges in non-canonical
                // form (e.g. `[1,10]` for int4range, which MZ stores as
                // `[1,11)`); without canonicalization those rows would not
                // compare or hash equal to MZ-constructed values.
                packer
                    .push_range(Range::new(Some((lower_bound, upper_bound))))
                    .context("pack range")?;

                return Ok(());
            }
            ColReader::IntervalYearMonth(array) => array
                .is_valid(idx)
                .then(|| array.value(idx))
                .map(|months| Datum::Interval(Interval::new(months, 0, 0))),
            ColReader::IntervalDayTime(array) => {
                array.is_valid(idx).then(|| array.value(idx)).map(|v| {
                    let micros = i64::from(v.milliseconds) * 1_000;
                    Datum::Interval(Interval::new(0, v.days, micros))
                })
            }
            ColReader::IntervalMonthDayNano(array) => {
                array.is_valid(idx).then(|| array.value(idx)).map(|v| {
                    let micros = v.nanoseconds / 1_000;
                    Datum::Interval(Interval::new(v.months, v.days, micros))
                })
            }
        };

        match datum {
            Some(d) => packer.push(d),
            None => packer.push(Datum::Null),
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use arrow::datatypes::Field;
    use mz_ore::collections::CollectionExt;

    use super::*;

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn smoketest_reader() {
        let desc = RelationDesc::builder()
            .with_column("bool", SqlScalarType::Bool.nullable(true))
            .with_column("int4", SqlScalarType::Int32.nullable(true))
            .with_column("uint8", SqlScalarType::UInt64.nullable(true))
            .with_column("float32", SqlScalarType::Float32.nullable(true))
            .with_column("string", SqlScalarType::String.nullable(true))
            .with_column("bytes", SqlScalarType::Bytes.nullable(true))
            .with_column("uuid", SqlScalarType::Uuid.nullable(true))
            .with_column("json", SqlScalarType::Jsonb.nullable(true))
            .with_column(
                "list",
                SqlScalarType::List {
                    element_type: Box::new(SqlScalarType::UInt32),
                    custom_id: None,
                }
                .nullable(true),
            )
            .finish();

        let mut og_row = Row::default();
        let mut packer = og_row.packer();

        packer.extend([
            Datum::True,
            Datum::Int32(42),
            Datum::UInt64(10000),
            Datum::Float32(OrderedFloat::from(-1.1f32)),
            Datum::String("hello world"),
            Datum::Bytes(b"1010101"),
            Datum::Uuid(uuid::Uuid::new_v4()),
        ]);
        JsonbPacker::new(&mut packer)
            .pack_serde_json(
                serde_json::json!({"code": 200, "email": "space_monkey@materialize.com"}),
            )
            .expect("failed to pack JSON");
        packer.push_list([Datum::UInt32(200), Datum::UInt32(300)]);

        let null_row = Row::pack(vec![Datum::Null; 9]);

        // Encode our data with our ArrowBuilder.
        let mut builder = crate::builder::ArrowBuilder::new(&desc, 2, 46).unwrap();
        builder.add_row(&og_row).unwrap();
        builder.add_row(&null_row).unwrap();
        let record_batch = builder.to_record_batch().unwrap();

        // Decode our data!
        let reader =
            ArrowReader::new(&desc, arrow::array::StructArray::from(record_batch)).unwrap();
        let mut rnd_row = Row::default();

        reader.read(0, &mut rnd_row).unwrap();
        assert_eq!(&og_row, &rnd_row);

        // Create a packer to clear the row alloc.
        rnd_row.packer();

        reader.read(1, &mut rnd_row).unwrap();
        assert_eq!(&null_row, &rnd_row);
    }

    /// Regression for SS-341: `COPY TO PARQUET` can write these scalar types,
    /// so `COPY FROM PARQUET` must read them back. Each must survive a
    /// builder -> reader round-trip.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn smoketest_extra_scalar_types() {
        let desc = RelationDesc::builder()
            .with_column("oid", SqlScalarType::Oid.nullable(true))
            .with_column("time", SqlScalarType::Time.nullable(true))
            .with_column(
                "timestamptz",
                SqlScalarType::TimestampTz { precision: None }.nullable(true),
            )
            .with_column("char", SqlScalarType::Char { length: None }.nullable(true))
            .with_column(
                "varchar",
                SqlScalarType::VarChar { max_length: None }.nullable(true),
            )
            .with_column("mz_timestamp", SqlScalarType::MzTimestamp.nullable(true))
            .finish();

        let tstz = CheckedTimestamp::from_timestamplike(
            DateTime::from_timestamp(1_600_000_000, 0).expect("valid timestamp"),
        )
        .expect("valid CheckedTimestamp");

        let og_row = Row::pack(vec![
            Datum::UInt32(42),
            // Sub-second component at microsecond resolution: the writer stores
            // Time64 microseconds, so nanoseconds would not round-trip.
            Datum::Time(NaiveTime::from_hms_micro_opt(12, 34, 56, 789_012).unwrap()),
            Datum::TimestampTz(tstz),
            Datum::String("abc"),
            Datum::String("hello world"),
            Datum::MzTimestamp(mz_repr::Timestamp::from(123_456_u64)),
        ]);
        let null_row = Row::pack(vec![Datum::Null; 6]);

        let mut builder = crate::builder::ArrowBuilder::new(&desc, 2, 64).unwrap();
        builder.add_row(&og_row).unwrap();
        builder.add_row(&null_row).unwrap();
        let record_batch = builder.to_record_batch().unwrap();

        let reader = ArrowReader::new(&desc, StructArray::from(record_batch)).unwrap();

        let mut got = Row::default();
        reader.read(0, &mut got).unwrap();
        assert_eq!(&og_row, &got, "values did not round-trip");

        got.packer();
        reader.read(1, &mut got).unwrap();
        assert_eq!(&null_row, &got, "NULLs did not round-trip");
    }

    /// Regression: an array column must survive a builder -> reader round-trip.
    /// The builder encodes an array as a struct of `{items, dimensions}`; the
    /// reader must reverse that back into a `Datum::Array`.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn smoketest_array() {
        let desc = RelationDesc::builder()
            .with_column(
                "arr",
                SqlScalarType::Array(Box::new(SqlScalarType::Int32)).nullable(true),
            )
            .finish();

        let mut row_1d = Row::default();
        row_1d
            .packer()
            .try_push_array(
                &[ArrayDimension {
                    lower_bound: 1,
                    length: 3,
                }],
                [Datum::Int32(1), Datum::Null, Datum::Int32(3)],
            )
            .unwrap();

        let mut row_empty = Row::default();
        row_empty
            .packer()
            .try_push_array(&[], std::iter::empty::<Datum>())
            .unwrap();

        let row_null = Row::pack(vec![Datum::Null]);

        // Encode with the builder, decode with the reader.
        let mut builder = crate::builder::ArrowBuilder::new(&desc, 3, 128).unwrap();
        builder.add_row(&row_1d).unwrap();
        builder.add_row(&row_empty).unwrap();
        builder.add_row(&row_null).unwrap();
        let record_batch = builder.to_record_batch().unwrap();

        let reader = ArrowReader::new(&desc, StructArray::from(record_batch)).unwrap();
        let mut got = Row::default();

        reader.read(0, &mut got).unwrap();
        assert_eq!(got, row_1d, "1-D array did not round-trip");

        got.packer();
        reader.read(1, &mut got).unwrap();
        assert_eq!(got, row_empty, "empty array did not round-trip");

        got.packer();
        reader.read(2, &mut got).unwrap();
        assert_eq!(got, row_null, "NULL array did not round-trip");
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn smoketest_decimal128() {
        let desc = RelationDesc::builder()
            .with_column(
                "a",
                SqlScalarType::Numeric { max_scale: None }.nullable(true),
            )
            .finish();

        let mut dec128 = arrow::array::Decimal128Builder::new();
        dec128 = dec128.with_precision_and_scale(12, 3).unwrap();

        // 1.234
        dec128.append_value(1234);
        dec128.append_null();
        // 100000000.009
        dec128.append_value(100000000009);

        let dec128 = dec128.finish();
        #[allow(clippy::as_conversions)]
        let batch = StructArray::from(vec![(
            Arc::new(Field::new("a", dec128.data_type().clone(), true)),
            Arc::new(dec128) as arrow::array::ArrayRef,
        )]);

        // Decode our data!
        let reader = ArrowReader::new(&desc, batch).unwrap();
        let mut rnd_row = Row::default();

        reader.read(0, &mut rnd_row).unwrap();
        let num = rnd_row.into_element().unwrap_numeric();
        assert_eq!(num.0, Numeric::from(1.234f64));

        // Create a packer to clear the row alloc.
        rnd_row.packer();

        reader.read(1, &mut rnd_row).unwrap();
        let num = rnd_row.into_element();
        assert_eq!(num, Datum::Null);

        // Create a packer to clear the row alloc.
        rnd_row.packer();

        reader.read(2, &mut rnd_row).unwrap();
        let num = rnd_row.into_element().unwrap_numeric();
        assert_eq!(num.0, Numeric::from(100000000.009f64));
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn smoketest_decimal256() {
        let desc = RelationDesc::builder()
            .with_column(
                "a",
                SqlScalarType::Numeric { max_scale: None }.nullable(true),
            )
            .finish();

        let mut dec256 = arrow::array::Decimal256Builder::new();
        dec256 = dec256.with_precision_and_scale(12, 3).unwrap();

        // 1.234
        dec256.append_value(arrow::datatypes::i256::from(1234));
        dec256.append_null();
        // 100000000.009
        dec256.append_value(arrow::datatypes::i256::from(100000000009i64));

        let dec256 = dec256.finish();
        #[allow(clippy::as_conversions)]
        let batch = StructArray::from(vec![(
            Arc::new(Field::new("a", dec256.data_type().clone(), true)),
            Arc::new(dec256) as arrow::array::ArrayRef,
        )]);

        // Decode our data!
        let reader = ArrowReader::new(&desc, batch).unwrap();
        let mut rnd_row = Row::default();

        reader.read(0, &mut rnd_row).unwrap();
        let num = rnd_row.into_element().unwrap_numeric();
        assert_eq!(num.0, Numeric::from(1.234f64));

        // Create a packer to clear the row alloc.
        rnd_row.packer();

        reader.read(1, &mut rnd_row).unwrap();
        let num = rnd_row.into_element();
        assert_eq!(num, Datum::Null);

        // Create a packer to clear the row alloc.
        rnd_row.packer();

        reader.read(2, &mut rnd_row).unwrap();
        let num = rnd_row.into_element().unwrap_numeric();
        assert_eq!(num.0, Numeric::from(100000000.009f64));
    }

    /// Regression test for SS-193: when the destination column declares a
    /// `max_scale`, the reader should round the decoded value to that scale
    /// rather than preserving the source file's scale.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn decimal_applies_destination_max_scale() {
        use mz_repr::adt::numeric::NumericMaxScale;

        // Destination column is numeric(10, 2): scale 2.
        let desc = RelationDesc::builder()
            .with_column(
                "a",
                SqlScalarType::Numeric {
                    max_scale: Some(NumericMaxScale::try_from(2_i64).unwrap()),
                }
                .nullable(true),
            )
            .finish();

        let expected = Numeric::from(10.45f64);

        // The source files carry scale-3 values (10.447) that don't round
        // evenly to the destination's scale 2.
        let mut dec128 = arrow::array::Decimal128Builder::new();
        dec128 = dec128.with_precision_and_scale(12, 3).unwrap();
        dec128.append_value(10447);
        let dec128 = dec128.finish();
        #[allow(clippy::as_conversions)]
        let batch128 = StructArray::from(vec![(
            Arc::new(Field::new("a", dec128.data_type().clone(), true)),
            Arc::new(dec128) as arrow::array::ArrayRef,
        )]);

        let reader = ArrowReader::new(&desc, batch128).unwrap();
        let mut rnd_row = Row::default();
        reader.read(0, &mut rnd_row).unwrap();
        let num = rnd_row.into_element().unwrap_numeric();
        assert_eq!(num.0, expected, "Decimal128 did not round to max_scale");

        let mut dec256 = arrow::array::Decimal256Builder::new();
        dec256 = dec256.with_precision_and_scale(12, 3).unwrap();
        dec256.append_value(arrow::datatypes::i256::from(10447));
        let dec256 = dec256.finish();
        #[allow(clippy::as_conversions)]
        let batch256 = StructArray::from(vec![(
            Arc::new(Field::new("a", dec256.data_type().clone(), true)),
            Arc::new(dec256) as arrow::array::ArrayRef,
        )]);

        let reader = ArrowReader::new(&desc, batch256).unwrap();
        let mut rnd_row = Row::default();
        reader.read(0, &mut rnd_row).unwrap();
        let num = rnd_row.into_element().unwrap_numeric();
        assert_eq!(num.0, expected, "Decimal256 did not round to max_scale");
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn decimal_rescale_to_higher_scale_overflows() {
        use mz_repr::adt::numeric::NumericMaxScale;

        let value: i128 = 10_i128.pow(33) - 1;

        let build_batch = || {
            let mut dec128 = arrow::array::Decimal128Builder::new();
            dec128 = dec128.with_precision_and_scale(38, 0).unwrap();
            dec128.append_value(value);
            let dec128 = dec128.finish();
            #[allow(clippy::as_conversions)]
            StructArray::from(vec![(
                Arc::new(Field::new("a", dec128.data_type().clone(), true)),
                Arc::new(dec128) as arrow::array::ArrayRef,
            )])
        };

        let desc_with_scale = |scale: i64| {
            RelationDesc::builder()
                .with_column(
                    "a",
                    SqlScalarType::Numeric {
                        max_scale: Some(NumericMaxScale::try_from(scale).unwrap()),
                    }
                    .nullable(true),
                )
                .finish()
        };

        // Test working case with supportable scale then broken case.
        let reader = ArrowReader::new(&desc_with_scale(2), build_batch()).unwrap();
        let mut row = Row::default();
        reader
            .read(0, &mut row)
            .expect("value must decode at destination scale 2");
        let num = row.into_element().unwrap_numeric();
        let mut expected = Numeric::try_from(value).unwrap();
        rescale(&mut expected, 2).unwrap();
        assert_eq!(num.0, expected, "value did not rescale to scale 2");

        let reader = ArrowReader::new(&desc_with_scale(8), build_batch()).unwrap();
        let mut row = Row::default();
        let err = reader
            .read(0, &mut row)
            .expect_err("value must overflow at destination scale 8");

        assert!(
            format!("{err:#}").contains("exceed maximum precision"),
            "unexpected error: {err:#}",
        );
    }

    /// Regression test for database-issues#11330: when a Parquet file authored
    /// by an external engine encodes a discrete range in non-canonical form
    /// (e.g. `[1,10]` for `int4range`), the reader must canonicalize it to MZ's
    /// internal form (`[1,11)`). Otherwise rows ingested via `COPY FROM PARQUET`
    /// don't compare or hash equal to logically-identical rows constructed
    /// inside MZ.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn range_canonicalizes_noncanonical_input() {
        use arrow::array::ArrayRef;
        use arrow::datatypes::DataType;
        use mz_repr::adt::range::{Range, RangeLowerBound, RangeUpperBound};

        let desc = RelationDesc::builder()
            .with_column(
                "r",
                SqlScalarType::Range {
                    element_type: Box::new(SqlScalarType::Int32),
                }
                .nullable(true),
            )
            .finish();

        // Build a range StructArray containing two non-canonical encodings:
        //   row 0: `[1,10]`  -> canonicalizes to `[1,11)`
        //   row 1: `(5,15)`  -> canonicalizes to `[6,15)`
        let lower = Int32Array::from(vec![Some(1), Some(5)]);
        let upper = Int32Array::from(vec![Some(10), Some(15)]);
        let lower_inclusive = BooleanArray::from(vec![true, false]);
        let upper_inclusive = BooleanArray::from(vec![true, false]);
        let empty = BooleanArray::from(vec![false, false]);

        #[allow(clippy::as_conversions)]
        let range_fields: Vec<(Arc<Field>, ArrayRef)> = vec![
            (
                Arc::new(Field::new("lower", DataType::Int32, true)),
                Arc::new(lower) as ArrayRef,
            ),
            (
                Arc::new(Field::new("upper", DataType::Int32, true)),
                Arc::new(upper) as ArrayRef,
            ),
            (
                Arc::new(Field::new("lower_inclusive", DataType::Boolean, false)),
                Arc::new(lower_inclusive) as ArrayRef,
            ),
            (
                Arc::new(Field::new("upper_inclusive", DataType::Boolean, false)),
                Arc::new(upper_inclusive) as ArrayRef,
            ),
            (
                Arc::new(Field::new("empty", DataType::Boolean, false)),
                Arc::new(empty) as ArrayRef,
            ),
        ];
        let range_struct = StructArray::from(range_fields);

        #[allow(clippy::as_conversions)]
        let batch = StructArray::from(vec![(
            Arc::new(Field::new("r", range_struct.data_type().clone(), true)),
            Arc::new(range_struct) as ArrayRef,
        )]);

        let reader = ArrowReader::new(&desc, batch).unwrap();

        // Row 0: `[1,10]` -> `[1,11)`
        let mut got = Row::default();
        reader.read(0, &mut got).unwrap();
        let mut want = Row::default();
        want.packer()
            .push_range(Range::new(Some((
                RangeLowerBound {
                    inclusive: true,
                    bound: Some(Datum::Int32(1)),
                },
                RangeUpperBound {
                    inclusive: true,
                    bound: Some(Datum::Int32(10)),
                },
            ))))
            .unwrap();
        assert_eq!(got, want, "row 0: [1,10] should canonicalize to [1,11)");

        // Row 1: `(5,15)` -> `[6,15)`
        let mut got = Row::default();
        reader.read(1, &mut got).unwrap();
        let mut want = Row::default();
        want.packer()
            .push_range(Range::new(Some((
                RangeLowerBound {
                    inclusive: false,
                    bound: Some(Datum::Int32(5)),
                },
                RangeUpperBound {
                    inclusive: false,
                    bound: Some(Datum::Int32(15)),
                },
            ))))
            .unwrap();
        assert_eq!(got, want, "row 1: (5,15) should canonicalize to [6,15)");
    }
}