Skip to main content

mz_arrow_util/
reader.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Reader for [`arrow`] data that outputs [`Row`]s.
11
12use std::sync::Arc;
13
14use anyhow::Context;
15use arrow::array::{
16    Array, BinaryArray, BinaryViewArray, BooleanArray, Date32Array, Date64Array, Decimal128Array,
17    Decimal256Array, FixedSizeBinaryArray, Float16Array, Float32Array, Float64Array, Int8Array,
18    Int16Array, Int32Array, Int64Array, IntervalDayTimeArray, IntervalMonthDayNanoArray,
19    IntervalYearMonthArray, LargeBinaryArray, LargeListArray, LargeStringArray, ListArray,
20    MapArray, StringArray, StringViewArray, StructArray, Time32MillisecondArray, Time32SecondArray,
21    TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
22    TimestampSecondArray, UInt8Array, UInt16Array, UInt32Array, UInt64Array,
23};
24use arrow::buffer::{NullBuffer, OffsetBuffer};
25use arrow::datatypes::{DataType, IntervalUnit, TimeUnit};
26use chrono::{DateTime, NaiveTime};
27use dec::OrderedDecimal;
28use itertools::Itertools;
29use mz_ore::cast::CastFrom;
30use mz_repr::adt::array::ArrayDimension;
31use mz_repr::adt::date::Date;
32use mz_repr::adt::interval::Interval;
33use mz_repr::adt::jsonb::JsonbPacker;
34use mz_repr::adt::numeric::{Numeric, rescale};
35use mz_repr::adt::range::{Range, RangeLowerBound, RangeUpperBound};
36use mz_repr::adt::timestamp::CheckedTimestamp;
37use mz_repr::{Datum, RelationDesc, Row, RowPacker, SharedRow, SqlScalarType};
38use ordered_float::OrderedFloat;
39use uuid::Uuid;
40
41use crate::mask_nulls;
42
43/// Type that can read out of an [`arrow::array::StructArray`] and into a [`Row`], given a
44/// [`RelationDesc`].
45///
46/// The inverse of a [`crate::builder::ArrowBuilder`].
47///
48/// Note: When creating an [`ArrowReader`] we perform a "one-time downcast" of the children Arrays
49/// from the [`StructArray`], into `enum ColReader`s. This is a much more verbose approach than the
50/// alternative of downcasting from a `dyn arrow::array::Array` every time we read a [`Row`], but
51/// it is _much_ more performant.
52pub struct ArrowReader {
53    len: usize,
54    readers: Vec<ColReader>,
55}
56
57impl ArrowReader {
58    /// Create an [`ArrowReader`] validating that the provided [`RelationDesc`] and [`StructArray`]
59    /// have a matching schema.
60    ///
61    /// The [`RelationDesc`] and [`StructArray`] need to uphold the following to be a valid pair:
62    ///
63    /// * Same number of columns.
64    /// * Columns of all the same name.
65    /// * Columns of compatible types.
66    ///
67    /// TODO(cf2): Relax some of these restrictions by allowing users to map column names, omit
68    /// columns, perform some lightweight casting, and matching not on column name but column
69    /// position.
70    /// TODO(cf2): Allow specifying an optional `arrow::Schema` for extra metadata.
71    pub fn new(desc: &RelationDesc, array: StructArray) -> Result<Self, anyhow::Error> {
72        let inner_columns = array.columns();
73        let desc_columns = desc.typ().columns();
74
75        if inner_columns.len() != desc_columns.len() {
76            return Err(anyhow::anyhow!(
77                "wrong number of columns {} vs {}",
78                inner_columns.len(),
79                desc_columns.len()
80            ));
81        }
82
83        let mut readers = Vec::with_capacity(desc_columns.len());
84        for (col_name, col_type) in desc.iter() {
85            let column = array
86                .column_by_name(col_name)
87                .ok_or_else(|| anyhow::anyhow!("'{col_name}' not found"))?;
88            let reader = scalar_type_and_array_to_reader(&col_type.scalar_type, Arc::clone(column))
89                .context(col_name.clone())?;
90
91            readers.push(reader);
92        }
93
94        Ok(ArrowReader {
95            len: array.len(),
96            readers,
97        })
98    }
99
100    /// Read the value at `idx` into the provided `Row`.
101    pub fn read(&self, idx: usize, row: &mut Row) -> Result<(), anyhow::Error> {
102        let mut packer = row.packer();
103        for reader in &self.readers {
104            reader.read(idx, &mut packer).context(idx)?;
105        }
106        Ok(())
107    }
108
109    /// Read all of the values in this [`ArrowReader`] into `rows`.
110    pub fn read_all(&self, rows: &mut Vec<Row>) -> Result<usize, anyhow::Error> {
111        for idx in 0..self.len {
112            let mut row = Row::default();
113            self.read(idx, &mut row).context(idx)?;
114            rows.push(row);
115        }
116        Ok(self.len)
117    }
118}
119
120fn scalar_type_and_array_to_reader(
121    scalar_type: &SqlScalarType,
122    array: Arc<dyn Array>,
123) -> Result<ColReader, anyhow::Error> {
124    fn downcast_array<T: arrow::array::Array + Clone + 'static>(array: Arc<dyn Array>) -> T {
125        array
126            .as_any()
127            .downcast_ref::<T>()
128            .expect("checked DataType")
129            .clone()
130    }
131
132    match (scalar_type, array.data_type()) {
133        (SqlScalarType::Bool, DataType::Boolean) => {
134            Ok(ColReader::Boolean(downcast_array::<BooleanArray>(array)))
135        }
136        (SqlScalarType::Int16 | SqlScalarType::Int32 | SqlScalarType::Int64, DataType::Int8) => {
137            let array = downcast_array::<Int8Array>(array);
138            let cast: fn(i8) -> Datum<'static> = match scalar_type {
139                SqlScalarType::Int16 => |x| Datum::Int16(i16::cast_from(x)),
140                SqlScalarType::Int32 => |x| Datum::Int32(i32::cast_from(x)),
141                SqlScalarType::Int64 => |x| Datum::Int64(i64::cast_from(x)),
142                _ => unreachable!("checked above"),
143            };
144            Ok(ColReader::Int8 { array, cast })
145        }
146        (SqlScalarType::Int16, DataType::Int16) => {
147            Ok(ColReader::Int16(downcast_array::<Int16Array>(array)))
148        }
149        (SqlScalarType::Int32, DataType::Int32) => {
150            Ok(ColReader::Int32(downcast_array::<Int32Array>(array)))
151        }
152        (SqlScalarType::Int64, DataType::Int64) => {
153            Ok(ColReader::Int64(downcast_array::<Int64Array>(array)))
154        }
155        (
156            SqlScalarType::UInt16 | SqlScalarType::UInt32 | SqlScalarType::UInt64,
157            DataType::UInt8,
158        ) => {
159            let array = downcast_array::<UInt8Array>(array);
160            let cast: fn(u8) -> Datum<'static> = match scalar_type {
161                SqlScalarType::UInt16 => |x| Datum::UInt16(u16::cast_from(x)),
162                SqlScalarType::UInt32 => |x| Datum::UInt32(u32::cast_from(x)),
163                SqlScalarType::UInt64 => |x| Datum::UInt64(u64::cast_from(x)),
164                _ => unreachable!("checked above"),
165            };
166            Ok(ColReader::UInt8 { array, cast })
167        }
168        (SqlScalarType::UInt16, DataType::UInt16) => {
169            Ok(ColReader::UInt16(downcast_array::<UInt16Array>(array)))
170        }
171        (SqlScalarType::UInt32, DataType::UInt32) => {
172            Ok(ColReader::UInt32(downcast_array::<UInt32Array>(array)))
173        }
174        (SqlScalarType::UInt64, DataType::UInt64) => {
175            Ok(ColReader::UInt64(downcast_array::<UInt64Array>(array)))
176        }
177        (SqlScalarType::Float32 | SqlScalarType::Float64, DataType::Float16) => {
178            let array = downcast_array::<Float16Array>(array);
179            let cast: fn(half::f16) -> Datum<'static> = match scalar_type {
180                SqlScalarType::Float32 => |x| Datum::Float32(OrderedFloat::from(x.to_f32())),
181                SqlScalarType::Float64 => |x| Datum::Float64(OrderedFloat::from(x.to_f64())),
182                _ => unreachable!("checked above"),
183            };
184            Ok(ColReader::Float16 { array, cast })
185        }
186        (SqlScalarType::Float32, DataType::Float32) => {
187            Ok(ColReader::Float32(downcast_array::<Float32Array>(array)))
188        }
189        (SqlScalarType::Float64, DataType::Float64) => {
190            Ok(ColReader::Float64(downcast_array::<Float64Array>(array)))
191        }
192        (SqlScalarType::Numeric { max_scale }, DataType::Decimal128(precision, scale)) => {
193            use num_traits::Pow;
194
195            let base = Numeric::from(10);
196            let scale = Numeric::from(*scale);
197            let scale_factor = base.pow(scale);
198
199            let precision = usize::cast_from(*precision);
200            // Don't use the context here, but make sure the precision is valid.
201            let mut ctx = dec::Context::<Numeric>::default();
202            ctx.set_precision(precision).map_err(|e| {
203                anyhow::anyhow!("invalid precision from Decimal128, {precision}, {e}")
204            })?;
205
206            let array = downcast_array::<Decimal128Array>(array);
207
208            Ok(ColReader::Decimal128 {
209                array,
210                scale_factor,
211                precision,
212                destination_max_scale: (*max_scale).map(|s| s.into_u8()),
213            })
214        }
215        (SqlScalarType::Numeric { max_scale }, DataType::Decimal256(precision, scale)) => {
216            use num_traits::Pow;
217
218            let base = Numeric::from(10);
219            let scale = Numeric::from(*scale);
220            let scale_factor = base.pow(scale);
221
222            let precision = usize::cast_from(*precision);
223            // Don't use the context here, but make sure the precision is valid.
224            let mut ctx = dec::Context::<Numeric>::default();
225            ctx.set_precision(precision).map_err(|e| {
226                anyhow::anyhow!("invalid precision from Decimal256, {precision}, {e}")
227            })?;
228
229            let array = downcast_array::<Decimal256Array>(array);
230
231            Ok(ColReader::Decimal256 {
232                array,
233                scale_factor,
234                precision,
235                destination_max_scale: (*max_scale).map(|s| s.into_u8()),
236            })
237        }
238        (SqlScalarType::Bytes, DataType::Binary) => {
239            Ok(ColReader::Binary(downcast_array::<BinaryArray>(array)))
240        }
241        (SqlScalarType::Bytes, DataType::LargeBinary) => {
242            let array = downcast_array::<LargeBinaryArray>(array);
243            Ok(ColReader::LargeBinary(array))
244        }
245        (SqlScalarType::Bytes, DataType::FixedSizeBinary(_)) => {
246            let array = downcast_array::<FixedSizeBinaryArray>(array);
247            Ok(ColReader::FixedSizeBinary(array))
248        }
249        (SqlScalarType::Bytes, DataType::BinaryView) => {
250            let array = downcast_array::<BinaryViewArray>(array);
251            Ok(ColReader::BinaryView(array))
252        }
253        (
254            SqlScalarType::Uuid,
255            DataType::Binary
256            | DataType::BinaryView
257            | DataType::LargeBinary
258            | DataType::FixedSizeBinary(_),
259        ) => {
260            let reader = scalar_type_and_array_to_reader(&SqlScalarType::Bytes, array)
261                .context("uuid reader")?;
262            Ok(ColReader::Uuid(Box::new(reader)))
263        }
264        (SqlScalarType::String, DataType::Utf8) => {
265            Ok(ColReader::String(downcast_array::<StringArray>(array)))
266        }
267        (SqlScalarType::String, DataType::LargeUtf8) => {
268            let array = downcast_array::<LargeStringArray>(array);
269            Ok(ColReader::LargeString(array))
270        }
271        (SqlScalarType::String, DataType::Utf8View) => {
272            let array = downcast_array::<StringViewArray>(array);
273            Ok(ColReader::StringView(array))
274        }
275        (SqlScalarType::Jsonb, DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View) => {
276            let reader = scalar_type_and_array_to_reader(&SqlScalarType::String, array)
277                .context("json reader")?;
278            Ok(ColReader::Jsonb(Box::new(reader)))
279        }
280        (SqlScalarType::Timestamp { .. }, DataType::Timestamp(TimeUnit::Second, None)) => {
281            let array = downcast_array::<TimestampSecondArray>(array);
282            Ok(ColReader::TimestampSecond(array))
283        }
284        (SqlScalarType::Timestamp { .. }, DataType::Timestamp(TimeUnit::Millisecond, None)) => {
285            let array = downcast_array::<TimestampMillisecondArray>(array);
286            Ok(ColReader::TimestampMillisecond(array))
287        }
288        (SqlScalarType::Timestamp { .. }, DataType::Timestamp(TimeUnit::Microsecond, None)) => {
289            let array = downcast_array::<TimestampMicrosecondArray>(array);
290            Ok(ColReader::TimestampMicrosecond(array))
291        }
292        (SqlScalarType::Timestamp { .. }, DataType::Timestamp(TimeUnit::Nanosecond, None)) => {
293            let array = downcast_array::<TimestampNanosecondArray>(array);
294            Ok(ColReader::TimestampNanosecond(array))
295        }
296        (SqlScalarType::Date, DataType::Date32) => {
297            let array = downcast_array::<Date32Array>(array);
298            Ok(ColReader::Date32(array))
299        }
300        (SqlScalarType::Date, DataType::Date64) => {
301            let array = downcast_array::<Date64Array>(array);
302            Ok(ColReader::Date64(array))
303        }
304        (SqlScalarType::Time, DataType::Time32(TimeUnit::Second)) => {
305            let array = downcast_array::<Time32SecondArray>(array);
306            Ok(ColReader::Time32Seconds(array))
307        }
308        (SqlScalarType::Time, DataType::Time32(TimeUnit::Millisecond)) => {
309            let array = downcast_array::<Time32MillisecondArray>(array);
310            Ok(ColReader::Time32Milliseconds(array))
311        }
312        (
313            SqlScalarType::List {
314                element_type,
315                custom_id: _,
316            },
317            DataType::List(_),
318        ) => {
319            let array = downcast_array::<ListArray>(array);
320            let inner_decoder =
321                scalar_type_and_array_to_reader(element_type, Arc::clone(array.values()))
322                    .context("list")?;
323            Ok(ColReader::List {
324                offsets: array.offsets().clone(),
325                values: Box::new(inner_decoder),
326                nulls: array.nulls().cloned(),
327            })
328        }
329        (
330            SqlScalarType::List {
331                element_type,
332                custom_id: _,
333            },
334            DataType::LargeList(_),
335        ) => {
336            let array = downcast_array::<LargeListArray>(array);
337            let inner_decoder =
338                scalar_type_and_array_to_reader(element_type, Arc::clone(array.values()))
339                    .context("large list")?;
340            Ok(ColReader::LargeList {
341                offsets: array.offsets().clone(),
342                values: Box::new(inner_decoder),
343                nulls: array.nulls().cloned(),
344            })
345        }
346        (SqlScalarType::Array(element_type), DataType::Struct(_)) => {
347            // The builder encodes an array as a struct of an `items` list (the
348            // flat, row-major elements) and a `dimensions` count. Reverse that.
349            let struct_array = downcast_array::<StructArray>(array);
350
351            let items = struct_array
352                .column_by_name("items")
353                .ok_or_else(|| anyhow::anyhow!("array struct missing 'items' field"))?;
354            let items = items
355                .as_any()
356                .downcast_ref::<ListArray>()
357                .ok_or_else(|| anyhow::anyhow!("array 'items' field is not a List"))?;
358            let values = scalar_type_and_array_to_reader(element_type, Arc::clone(items.values()))
359                .context("array items")?;
360
361            let dims_col = struct_array
362                .column_by_name("dimensions")
363                .ok_or_else(|| anyhow::anyhow!("array struct missing 'dimensions' field"))?;
364            // The builder writes `dimensions` as UInt8. An Iceberg round-trip
365            // widens it to Int32 (Iceberg has no narrow integer types), so
366            // accept both.
367            let dims = match dims_col.data_type() {
368                DataType::UInt8 => {
369                    ArrayDims::UInt8(downcast_array::<UInt8Array>(Arc::clone(dims_col)))
370                }
371                DataType::Int32 => {
372                    ArrayDims::Int32(downcast_array::<Int32Array>(Arc::clone(dims_col)))
373                }
374                other => anyhow::bail!("unsupported array 'dimensions' type: {other:?}"),
375            };
376
377            Ok(ColReader::Array {
378                offsets: items.offsets().clone(),
379                values: Box::new(values),
380                dims,
381                nulls: struct_array.nulls().cloned(),
382            })
383        }
384        (
385            SqlScalarType::Record {
386                fields,
387                custom_id: _,
388            },
389            DataType::Struct(_),
390        ) => {
391            let record_array = downcast_array::<StructArray>(array);
392            let null_mask = record_array.nulls();
393
394            let mut decoders = Vec::with_capacity(fields.len());
395            for (name, typ) in fields.iter() {
396                let inner_array = record_array
397                    .column_by_name(name)
398                    .ok_or_else(|| anyhow::anyhow!("missing name '{name}'"))?;
399                let inner_array = mask_nulls(inner_array, null_mask);
400                let inner_decoder = scalar_type_and_array_to_reader(&typ.scalar_type, inner_array)
401                    .context(name.clone())?;
402
403                decoders.push(Box::new(inner_decoder));
404            }
405
406            Ok(ColReader::Record {
407                fields: decoders,
408                nulls: null_mask.cloned(),
409            })
410        }
411        (
412            SqlScalarType::Map {
413                value_type,
414                custom_id: _,
415            },
416            DataType::Map(_, _),
417        ) => {
418            let map_array = downcast_array::<MapArray>(array);
419            let keys = map_array
420                .keys()
421                .as_any()
422                .downcast_ref::<StringArray>()
423                .expect("map keys should be Utf8 strings")
424                .clone();
425            let values_reader =
426                scalar_type_and_array_to_reader(value_type, Arc::clone(map_array.values()))
427                    .context("map values")?;
428            Ok(ColReader::Map {
429                offsets: map_array.offsets().clone(),
430                keys,
431                values: Box::new(values_reader),
432                nulls: map_array.nulls().cloned(),
433            })
434        }
435        (SqlScalarType::Range { element_type }, DataType::Struct(_)) => {
436            let struct_array = downcast_array::<StructArray>(array);
437            let lower_array = struct_array
438                .column_by_name("lower")
439                .ok_or_else(|| anyhow::anyhow!("range struct missing 'lower' field"))?;
440            let upper_array = struct_array
441                .column_by_name("upper")
442                .ok_or_else(|| anyhow::anyhow!("range struct missing 'upper' field"))?;
443            let empty_array = struct_array
444                .column_by_name("empty")
445                .ok_or_else(|| anyhow::anyhow!("range struct missing 'empty' field"))?;
446            let lower_inclusive_array = struct_array
447                .column_by_name("lower_inclusive")
448                .ok_or_else(|| anyhow::anyhow!("range struct missing 'lower_inclusive' field"))?;
449            let upper_inclusive_array = struct_array
450                .column_by_name("upper_inclusive")
451                .ok_or_else(|| anyhow::anyhow!("range struct missing 'upper_inclusive' field"))?;
452
453            let lower_reader =
454                scalar_type_and_array_to_reader(element_type, Arc::clone(lower_array))
455                    .context("range lower")?;
456            let upper_reader =
457                scalar_type_and_array_to_reader(element_type, Arc::clone(upper_array))
458                    .context("range upper")?;
459
460            let empty = downcast_array::<BooleanArray>(Arc::clone(empty_array));
461            let lower_inclusive = downcast_array::<BooleanArray>(Arc::clone(lower_inclusive_array));
462            let upper_inclusive = downcast_array::<BooleanArray>(Arc::clone(upper_inclusive_array));
463
464            let lower_nulls = lower_array.nulls().cloned();
465            let upper_nulls = upper_array.nulls().cloned();
466
467            Ok(ColReader::Range {
468                lower: Box::new(lower_reader),
469                lower_nulls,
470                upper: Box::new(upper_reader),
471                upper_nulls,
472                lower_inclusive,
473                upper_inclusive,
474                empty,
475                nulls: struct_array.nulls().cloned(),
476            })
477        }
478        (SqlScalarType::Interval, DataType::Interval(IntervalUnit::YearMonth)) => {
479            Ok(ColReader::IntervalYearMonth(downcast_array::<
480                IntervalYearMonthArray,
481            >(array)))
482        }
483        (SqlScalarType::Interval, DataType::Interval(IntervalUnit::DayTime)) => {
484            Ok(ColReader::IntervalDayTime(downcast_array::<
485                IntervalDayTimeArray,
486            >(array)))
487        }
488        (SqlScalarType::Interval, DataType::Interval(IntervalUnit::MonthDayNano)) => {
489            Ok(ColReader::IntervalMonthDayNano(downcast_array::<
490                IntervalMonthDayNanoArray,
491            >(array)))
492        }
493        other => anyhow::bail!("unsupported: {other:?}"),
494    }
495}
496
497/// The `dimensions` field of an encoded array. The builder writes it as
498/// [`UInt8`](DataType::UInt8); an Iceberg round-trip widens it to
499/// [`Int32`](DataType::Int32).
500enum ArrayDims {
501    UInt8(UInt8Array),
502    Int32(Int32Array),
503}
504
505impl ArrayDims {
506    /// The number of dimensions of the array at `idx`.
507    fn ndims(&self, idx: usize) -> Result<u8, anyhow::Error> {
508        match self {
509            ArrayDims::UInt8(array) => Ok(array.value(idx)),
510            ArrayDims::Int32(array) => {
511                u8::try_from(array.value(idx)).context("array dimension count out of range")
512            }
513        }
514    }
515}
516
517/// A "downcasted" version of [`arrow::array::Array`] that supports reading [`Datum`]s.
518///
519/// Note: While this is fairly verbose, one-time "downcasting" to an enum is _much_ more performant
520/// than downcasting every time we read a [`Datum`].
521enum ColReader {
522    Boolean(arrow::array::BooleanArray),
523
524    Int8 {
525        array: arrow::array::Int8Array,
526        cast: fn(i8) -> Datum<'static>,
527    },
528    Int16(arrow::array::Int16Array),
529    Int32(arrow::array::Int32Array),
530    Int64(arrow::array::Int64Array),
531
532    UInt8 {
533        array: arrow::array::UInt8Array,
534        cast: fn(u8) -> Datum<'static>,
535    },
536    UInt16(arrow::array::UInt16Array),
537    UInt32(arrow::array::UInt32Array),
538    UInt64(arrow::array::UInt64Array),
539
540    Float16 {
541        array: arrow::array::Float16Array,
542        cast: fn(half::f16) -> Datum<'static>,
543    },
544    Float32(arrow::array::Float32Array),
545    Float64(arrow::array::Float64Array),
546
547    Decimal128 {
548        array: Decimal128Array,
549        scale_factor: Numeric,
550        precision: usize,
551        destination_max_scale: Option<u8>,
552    },
553    Decimal256 {
554        array: Decimal256Array,
555        scale_factor: Numeric,
556        precision: usize,
557        destination_max_scale: Option<u8>,
558    },
559
560    Binary(arrow::array::BinaryArray),
561    LargeBinary(arrow::array::LargeBinaryArray),
562    FixedSizeBinary(arrow::array::FixedSizeBinaryArray),
563    BinaryView(arrow::array::BinaryViewArray),
564    Uuid(Box<ColReader>),
565
566    String(arrow::array::StringArray),
567    LargeString(arrow::array::LargeStringArray),
568    StringView(arrow::array::StringViewArray),
569    Jsonb(Box<ColReader>),
570
571    TimestampSecond(arrow::array::TimestampSecondArray),
572    TimestampMillisecond(arrow::array::TimestampMillisecondArray),
573    TimestampMicrosecond(arrow::array::TimestampMicrosecondArray),
574    TimestampNanosecond(arrow::array::TimestampNanosecondArray),
575
576    Date32(Date32Array),
577    Date64(Date64Array),
578
579    Time32Seconds(Time32SecondArray),
580    Time32Milliseconds(arrow::array::Time32MillisecondArray),
581
582    List {
583        offsets: OffsetBuffer<i32>,
584        values: Box<ColReader>,
585        nulls: Option<NullBuffer>,
586    },
587    LargeList {
588        offsets: OffsetBuffer<i64>,
589        values: Box<ColReader>,
590        nulls: Option<NullBuffer>,
591    },
592
593    Array {
594        offsets: OffsetBuffer<i32>,
595        values: Box<ColReader>,
596        dims: ArrayDims,
597        nulls: Option<NullBuffer>,
598    },
599
600    Record {
601        fields: Vec<Box<ColReader>>,
602        nulls: Option<NullBuffer>,
603    },
604
605    Map {
606        offsets: OffsetBuffer<i32>,
607        keys: StringArray,
608        values: Box<ColReader>,
609        nulls: Option<NullBuffer>,
610    },
611
612    Range {
613        lower: Box<ColReader>,
614        lower_nulls: Option<NullBuffer>,
615        upper: Box<ColReader>,
616        upper_nulls: Option<NullBuffer>,
617        lower_inclusive: BooleanArray,
618        upper_inclusive: BooleanArray,
619        empty: BooleanArray,
620        nulls: Option<NullBuffer>,
621    },
622
623    IntervalYearMonth(IntervalYearMonthArray),
624    IntervalDayTime(IntervalDayTimeArray),
625    IntervalMonthDayNano(IntervalMonthDayNanoArray),
626}
627
628impl ColReader {
629    fn read(&self, idx: usize, packer: &mut RowPacker) -> Result<(), anyhow::Error> {
630        let datum = match self {
631            ColReader::Boolean(array) => array
632                .is_valid(idx)
633                .then(|| array.value(idx))
634                .map(|x| if x { Datum::True } else { Datum::False }),
635            ColReader::Int8 { array, cast } => {
636                array.is_valid(idx).then(|| array.value(idx)).map(cast)
637            }
638            ColReader::Int16(array) => array
639                .is_valid(idx)
640                .then(|| array.value(idx))
641                .map(Datum::Int16),
642            ColReader::Int32(array) => array
643                .is_valid(idx)
644                .then(|| array.value(idx))
645                .map(Datum::Int32),
646            ColReader::Int64(array) => array
647                .is_valid(idx)
648                .then(|| array.value(idx))
649                .map(Datum::Int64),
650            ColReader::UInt8 { array, cast } => {
651                array.is_valid(idx).then(|| array.value(idx)).map(cast)
652            }
653            ColReader::UInt16(array) => array
654                .is_valid(idx)
655                .then(|| array.value(idx))
656                .map(Datum::UInt16),
657            ColReader::UInt32(array) => array
658                .is_valid(idx)
659                .then(|| array.value(idx))
660                .map(Datum::UInt32),
661            ColReader::UInt64(array) => array
662                .is_valid(idx)
663                .then(|| array.value(idx))
664                .map(Datum::UInt64),
665            ColReader::Float16 { array, cast } => {
666                array.is_valid(idx).then(|| array.value(idx)).map(cast)
667            }
668            ColReader::Float32(array) => array
669                .is_valid(idx)
670                .then(|| array.value(idx))
671                .map(|x| Datum::Float32(OrderedFloat(x))),
672            ColReader::Float64(array) => array
673                .is_valid(idx)
674                .then(|| array.value(idx))
675                .map(|x| Datum::Float64(OrderedFloat(x))),
676            ColReader::Decimal128 {
677                array,
678                scale_factor,
679                precision,
680                destination_max_scale,
681            } => array
682                .is_valid(idx)
683                .then(|| array.value(idx))
684                .map(|x| {
685                    // Create a Numeric from our i128 with precision.
686                    let mut ctx = dec::Context::<Numeric>::default();
687                    ctx.set_precision(*precision).expect("checked before");
688                    let mut num = ctx.from_i128(x);
689
690                    // Scale the number.
691                    ctx.div(&mut num, scale_factor);
692
693                    if let Some(destination_max_scale) = destination_max_scale {
694                        rescale(&mut num, *destination_max_scale)?;
695                    }
696
697                    Ok::<_, anyhow::Error>(Datum::Numeric(OrderedDecimal(num)))
698                })
699                .transpose()?,
700            ColReader::Decimal256 {
701                array,
702                scale_factor,
703                precision,
704                destination_max_scale,
705            } => array
706                .is_valid(idx)
707                .then(|| array.value(idx))
708                .map(|x| {
709                    let s = x.to_string();
710
711                    // Parse a i256 from it's String representation.
712                    //
713                    // TODO(cf3): See if we can add support for 256-bit numbers to the `dec` crate.
714                    let mut ctx = dec::Context::<Numeric>::default();
715                    ctx.set_precision(*precision).expect("checked before");
716                    let mut num = ctx
717                        .parse(s)
718                        .map_err(|e| anyhow::anyhow!("decimal out of range: {e}"))?;
719
720                    // Scale the number.
721                    ctx.div(&mut num, scale_factor);
722
723                    if let Some(destination_max_scale) = destination_max_scale {
724                        rescale(&mut num, *destination_max_scale)?;
725                    }
726
727                    Ok::<_, anyhow::Error>(Datum::Numeric(OrderedDecimal(num)))
728                })
729                .transpose()?,
730            ColReader::Binary(array) => array
731                .is_valid(idx)
732                .then(|| array.value(idx))
733                .map(Datum::Bytes),
734            ColReader::LargeBinary(array) => array
735                .is_valid(idx)
736                .then(|| array.value(idx))
737                .map(Datum::Bytes),
738            ColReader::FixedSizeBinary(array) => array
739                .is_valid(idx)
740                .then(|| array.value(idx))
741                .map(Datum::Bytes),
742            ColReader::BinaryView(array) => array
743                .is_valid(idx)
744                .then(|| array.value(idx))
745                .map(Datum::Bytes),
746            ColReader::Uuid(reader) => {
747                // First read a binary value into a temp row, and later parse that as UUID into our
748                // actual Row Packer.
749                let mut temp_row = SharedRow::get();
750                reader.read(idx, &mut temp_row.packer()).context("uuid")?;
751                let slice = match temp_row.unpack_first() {
752                    Datum::Bytes(slice) => slice,
753                    Datum::Null => {
754                        packer.push(Datum::Null);
755                        return Ok(());
756                    }
757                    other => anyhow::bail!("expected String, found {other:?}"),
758                };
759
760                let uuid = Uuid::from_slice(slice).context("parsing uuid")?;
761                Some(Datum::Uuid(uuid))
762            }
763            ColReader::String(array) => array
764                .is_valid(idx)
765                .then(|| array.value(idx))
766                .map(Datum::String),
767            ColReader::LargeString(array) => array
768                .is_valid(idx)
769                .then(|| array.value(idx))
770                .map(Datum::String),
771            ColReader::StringView(array) => array
772                .is_valid(idx)
773                .then(|| array.value(idx))
774                .map(Datum::String),
775            ColReader::Jsonb(reader) => {
776                // First read a string value into a temp row, and later parse that as JSON into our
777                // actual Row Packer.
778                let mut temp_row = SharedRow::get();
779                reader.read(idx, &mut temp_row.packer()).context("jsonb")?;
780                let value = match temp_row.unpack_first() {
781                    Datum::String(value) => value,
782                    Datum::Null => {
783                        packer.push(Datum::Null);
784                        return Ok(());
785                    }
786                    other => anyhow::bail!("expected String, found {other:?}"),
787                };
788
789                JsonbPacker::new(packer)
790                    .pack_str(value)
791                    .context("roundtrip json")?;
792
793                // Return early because we've already packed the necessasry Datums.
794                return Ok(());
795            }
796            ColReader::TimestampSecond(array) => array
797                .is_valid(idx)
798                .then(|| array.value(idx))
799                .map(|secs| {
800                    let dt = DateTime::from_timestamp(secs, 0)
801                        .ok_or_else(|| anyhow::anyhow!("invalid timestamp seconds {secs}"))?;
802                    let dt = CheckedTimestamp::from_timestamplike(dt.naive_utc())
803                        .context("TimestampSeconds")?;
804                    Ok::<_, anyhow::Error>(Datum::Timestamp(dt))
805                })
806                .transpose()?,
807            ColReader::TimestampMillisecond(array) => array
808                .is_valid(idx)
809                .then(|| array.value(idx))
810                .map(|millis| {
811                    let dt = DateTime::from_timestamp_millis(millis).ok_or_else(|| {
812                        anyhow::anyhow!("invalid timestamp milliseconds {millis}")
813                    })?;
814                    let dt = CheckedTimestamp::from_timestamplike(dt.naive_utc())
815                        .context("TimestampMillis")?;
816                    Ok::<_, anyhow::Error>(Datum::Timestamp(dt))
817                })
818                .transpose()?,
819            ColReader::TimestampMicrosecond(array) => array
820                .is_valid(idx)
821                .then(|| array.value(idx))
822                .map(|micros| {
823                    let dt = DateTime::from_timestamp_micros(micros).ok_or_else(|| {
824                        anyhow::anyhow!("invalid timestamp microseconds {micros}")
825                    })?;
826                    let dt = CheckedTimestamp::from_timestamplike(dt.naive_utc())
827                        .context("TimestampMicros")?;
828                    Ok::<_, anyhow::Error>(Datum::Timestamp(dt))
829                })
830                .transpose()?,
831            ColReader::TimestampNanosecond(array) => array
832                .is_valid(idx)
833                .then(|| array.value(idx))
834                .map(|nanos| {
835                    let dt = DateTime::from_timestamp_nanos(nanos);
836                    let dt = CheckedTimestamp::from_timestamplike(dt.naive_utc())
837                        .context("TimestampNanos")?;
838                    Ok::<_, anyhow::Error>(Datum::Timestamp(dt))
839                })
840                .transpose()?,
841            ColReader::Date32(array) => array
842                .is_valid(idx)
843                .then(|| array.value(idx))
844                .map(|unix_days| {
845                    let date = Date::from_unix_epoch(unix_days).context("date32")?;
846                    Ok::<_, anyhow::Error>(Datum::Date(date))
847                })
848                .transpose()?,
849            ColReader::Date64(array) => array
850                .is_valid(idx)
851                .then(|| array.value(idx))
852                .map(|unix_millis| {
853                    let date = DateTime::from_timestamp_millis(unix_millis)
854                        .ok_or_else(|| anyhow::anyhow!("invalid Date64 {unix_millis}"))?;
855                    let unix_epoch = DateTime::from_timestamp(0, 0)
856                        .expect("UNIX epoch")
857                        .date_naive();
858                    let delta = date.date_naive().signed_duration_since(unix_epoch);
859                    let days: i32 = delta.num_days().try_into().context("date64")?;
860                    let date = Date::from_unix_epoch(days).context("date64")?;
861                    Ok::<_, anyhow::Error>(Datum::Date(date))
862                })
863                .transpose()?,
864            ColReader::Time32Seconds(array) => array
865                .is_valid(idx)
866                .then(|| array.value(idx))
867                .map(|secs| {
868                    let usecs: u32 = secs.try_into().context("time32 seconds")?;
869                    let time = NaiveTime::from_num_seconds_from_midnight_opt(usecs, 0)
870                        .ok_or_else(|| anyhow::anyhow!("invalid Time32 Seconds {secs}"))?;
871                    Ok::<_, anyhow::Error>(Datum::Time(time))
872                })
873                .transpose()?,
874            ColReader::Time32Milliseconds(array) => array
875                .is_valid(idx)
876                .then(|| array.value(idx))
877                .map(|millis| {
878                    let umillis: u32 = millis.try_into().context("time32 milliseconds")?;
879                    let usecs = umillis / 1000;
880                    let unanos = (umillis % 1000).saturating_mul(1_000_000);
881                    let time = NaiveTime::from_num_seconds_from_midnight_opt(usecs, unanos)
882                        .ok_or_else(|| anyhow::anyhow!("invalid Time32 Milliseconds {umillis}"))?;
883                    Ok::<_, anyhow::Error>(Datum::Time(time))
884                })
885                .transpose()?,
886            ColReader::List {
887                offsets,
888                values,
889                nulls,
890            } => {
891                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
892                if !is_valid {
893                    packer.push(Datum::Null);
894                    return Ok(());
895                }
896
897                let start: usize = offsets[idx].try_into().context("list start offset")?;
898                let end: usize = offsets[idx + 1].try_into().context("list end offset")?;
899
900                packer
901                    .push_list_with(|packer| {
902                        for idx in start..end {
903                            values.read(idx, packer)?;
904                        }
905                        Ok::<_, anyhow::Error>(())
906                    })
907                    .context("pack list")?;
908
909                // Return early because we've already packed the necessasry Datums.
910                return Ok(());
911            }
912            ColReader::LargeList {
913                offsets,
914                values,
915                nulls,
916            } => {
917                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
918                if !is_valid {
919                    packer.push(Datum::Null);
920                    return Ok(());
921                }
922
923                let start: usize = offsets[idx].try_into().context("list start offset")?;
924                let end: usize = offsets[idx + 1].try_into().context("list end offset")?;
925
926                packer
927                    .push_list_with(|packer| {
928                        for idx in start..end {
929                            values.read(idx, packer)?;
930                        }
931                        Ok::<_, anyhow::Error>(())
932                    })
933                    .context("pack list")?;
934
935                // Return early because we've already packed the necessasry Datums.
936                return Ok(());
937            }
938            ColReader::Array {
939                offsets,
940                values,
941                dims,
942                nulls,
943            } => {
944                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
945                if !is_valid {
946                    packer.push(Datum::Null);
947                    return Ok(());
948                }
949
950                let start: usize = offsets[idx].try_into().context("array start offset")?;
951                let end: usize = offsets[idx + 1].try_into().context("array end offset")?;
952                let nelements = end - start;
953
954                // The encoding stores only the dimension count and the flat,
955                // row-major elements (see the builder), so per-dimension extents
956                // are recoverable only for 0- and 1-dimensional arrays. A
957                // higher-dimensional array's extents cannot be reconstructed, so
958                // reject it rather than guess a shape.
959                let ndims = dims.ndims(idx)?;
960                if ndims > 1 {
961                    anyhow::bail!(
962                        "cannot decode {ndims}-dimensional array from parquet: the encoding \
963                         records only the dimension count, not per-dimension extents"
964                    );
965                }
966                let one_dim = [ArrayDimension {
967                    lower_bound: 1,
968                    length: nelements,
969                }];
970                let array_dims: &[ArrayDimension] = if ndims == 0 { &[] } else { &one_dim };
971
972                // SAFETY: the closure returns exactly the number of elements it
973                // pushes (`end - start`).
974                unsafe {
975                    packer.push_array_with_unchecked(array_dims, |packer| {
976                        for idx in start..end {
977                            values.read(idx, packer)?;
978                        }
979                        Ok::<_, anyhow::Error>(end - start)
980                    })
981                }
982                .context("pack array")?;
983
984                // Return early because we've already packed the necessasry Datums.
985                return Ok(());
986            }
987            ColReader::Record { fields, nulls } => {
988                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
989                if !is_valid {
990                    packer.push(Datum::Null);
991                    return Ok(());
992                }
993
994                packer
995                    .push_list_with(|packer| {
996                        for field in fields {
997                            field.read(idx, packer)?;
998                        }
999                        Ok::<_, anyhow::Error>(())
1000                    })
1001                    .context("pack record")?;
1002
1003                // Return early because we've already packed the necessasry Datums.
1004                return Ok(());
1005            }
1006            ColReader::Map {
1007                offsets,
1008                keys,
1009                values,
1010                nulls,
1011            } => {
1012                let is_non_null = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
1013                if !is_non_null {
1014                    packer.push(Datum::Null);
1015                    return Ok(());
1016                }
1017
1018                let start: usize = offsets[idx].try_into().context("map start offset")?;
1019                let end: usize = offsets[idx + 1].try_into().context("map end offset")?;
1020
1021                // Arrow's MapArray doesn't guarantee that keys are in sorted order, but Materialize's
1022                // Datum::Map does, so we need to sort the keys here before packing them, or else
1023                // many assumptions will break.
1024                let mut kv_sorted = (start..end)
1025                    .map(|i| (keys.value(i), i))
1026                    .sorted_by_key(|(k, _)| *k)
1027                    .peekable();
1028
1029                packer
1030                    .push_dict_with(|packer| {
1031                        while let Some((key, i)) = kv_sorted.next() {
1032                            // Parquet docs state that if there are duplicate keys, the last value
1033                            // should be used, so skip duplicates here.
1034                            //
1035                            // sorted_by_key is a stable sort, so entries with duplicate keys will
1036                            // maintain their original order, and we can pick the last one here.
1037                            if let Some((next_key, _)) = kv_sorted.peek() {
1038                                if key == *next_key {
1039                                    continue;
1040                                }
1041                            }
1042                            packer.push(Datum::String(key));
1043                            values.read(i, packer)?;
1044                        }
1045                        Ok::<_, anyhow::Error>(())
1046                    })
1047                    .context("pack map")?;
1048
1049                // Return early because we've already packed the necessary Datums.
1050                return Ok(());
1051            }
1052            ColReader::Range {
1053                lower,
1054                lower_nulls,
1055                upper,
1056                upper_nulls,
1057                lower_inclusive,
1058                upper_inclusive,
1059                empty,
1060                nulls,
1061            } => {
1062                let is_valid = nulls.as_ref().map(|n| n.is_valid(idx)).unwrap_or(true);
1063                if !is_valid {
1064                    packer.push(Datum::Null);
1065                    return Ok(());
1066                }
1067
1068                if empty.value(idx) {
1069                    packer.push(Datum::Range(Range { inner: None }));
1070                    return Ok(());
1071                }
1072
1073                let lower_is_infinite = lower_nulls
1074                    .as_ref()
1075                    .map(|n| !n.is_valid(idx))
1076                    .unwrap_or(false);
1077                let upper_is_infinite = upper_nulls
1078                    .as_ref()
1079                    .map(|n| !n.is_valid(idx))
1080                    .unwrap_or(false);
1081
1082                // Read finite bounds into owned Rows that live for the rest of
1083                // this block, so the Datums we borrow out of them stay valid
1084                // for the `push_range` call below.
1085                let lower_row = if lower_is_infinite {
1086                    None
1087                } else {
1088                    let mut temp = SharedRow::get();
1089                    lower.read(idx, &mut temp.packer())?;
1090                    Some(temp.clone())
1091                };
1092                let upper_row = if upper_is_infinite {
1093                    None
1094                } else {
1095                    let mut temp = SharedRow::get();
1096                    upper.read(idx, &mut temp.packer())?;
1097                    Some(temp.clone())
1098                };
1099
1100                let lower_bound = RangeLowerBound {
1101                    inclusive: lower_inclusive.value(idx),
1102                    bound: lower_row.as_ref().map(|row| row.unpack_first()),
1103                };
1104                let upper_bound = RangeUpperBound {
1105                    inclusive: upper_inclusive.value(idx),
1106                    bound: upper_row.as_ref().map(|row| row.unpack_first()),
1107                };
1108
1109                // Use `push_range` (not `push_range_with`) so the range is
1110                // canonicalized before being packed. Parquet files authored by
1111                // external engines may encode discrete ranges in non-canonical
1112                // form (e.g. `[1,10]` for int4range, which MZ stores as
1113                // `[1,11)`); without canonicalization those rows would not
1114                // compare or hash equal to MZ-constructed values.
1115                packer
1116                    .push_range(Range::new(Some((lower_bound, upper_bound))))
1117                    .context("pack range")?;
1118
1119                return Ok(());
1120            }
1121            ColReader::IntervalYearMonth(array) => array
1122                .is_valid(idx)
1123                .then(|| array.value(idx))
1124                .map(|months| Datum::Interval(Interval::new(months, 0, 0))),
1125            ColReader::IntervalDayTime(array) => {
1126                array.is_valid(idx).then(|| array.value(idx)).map(|v| {
1127                    let micros = i64::from(v.milliseconds) * 1_000;
1128                    Datum::Interval(Interval::new(0, v.days, micros))
1129                })
1130            }
1131            ColReader::IntervalMonthDayNano(array) => {
1132                array.is_valid(idx).then(|| array.value(idx)).map(|v| {
1133                    let micros = v.nanoseconds / 1_000;
1134                    Datum::Interval(Interval::new(v.months, v.days, micros))
1135                })
1136            }
1137        };
1138
1139        match datum {
1140            Some(d) => packer.push(d),
1141            None => packer.push(Datum::Null),
1142        }
1143
1144        Ok(())
1145    }
1146}
1147
1148#[cfg(test)]
1149mod tests {
1150    use arrow::datatypes::Field;
1151    use mz_ore::collections::CollectionExt;
1152
1153    use super::*;
1154
1155    #[mz_ore::test]
1156    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
1157    fn smoketest_reader() {
1158        let desc = RelationDesc::builder()
1159            .with_column("bool", SqlScalarType::Bool.nullable(true))
1160            .with_column("int4", SqlScalarType::Int32.nullable(true))
1161            .with_column("uint8", SqlScalarType::UInt64.nullable(true))
1162            .with_column("float32", SqlScalarType::Float32.nullable(true))
1163            .with_column("string", SqlScalarType::String.nullable(true))
1164            .with_column("bytes", SqlScalarType::Bytes.nullable(true))
1165            .with_column("uuid", SqlScalarType::Uuid.nullable(true))
1166            .with_column("json", SqlScalarType::Jsonb.nullable(true))
1167            .with_column(
1168                "list",
1169                SqlScalarType::List {
1170                    element_type: Box::new(SqlScalarType::UInt32),
1171                    custom_id: None,
1172                }
1173                .nullable(true),
1174            )
1175            .finish();
1176
1177        let mut og_row = Row::default();
1178        let mut packer = og_row.packer();
1179
1180        packer.extend([
1181            Datum::True,
1182            Datum::Int32(42),
1183            Datum::UInt64(10000),
1184            Datum::Float32(OrderedFloat::from(-1.1f32)),
1185            Datum::String("hello world"),
1186            Datum::Bytes(b"1010101"),
1187            Datum::Uuid(uuid::Uuid::new_v4()),
1188        ]);
1189        JsonbPacker::new(&mut packer)
1190            .pack_serde_json(
1191                serde_json::json!({"code": 200, "email": "space_monkey@materialize.com"}),
1192            )
1193            .expect("failed to pack JSON");
1194        packer.push_list([Datum::UInt32(200), Datum::UInt32(300)]);
1195
1196        let null_row = Row::pack(vec![Datum::Null; 9]);
1197
1198        // Encode our data with our ArrowBuilder.
1199        let mut builder = crate::builder::ArrowBuilder::new(&desc, 2, 46).unwrap();
1200        builder.add_row(&og_row).unwrap();
1201        builder.add_row(&null_row).unwrap();
1202        let record_batch = builder.to_record_batch().unwrap();
1203
1204        // Decode our data!
1205        let reader =
1206            ArrowReader::new(&desc, arrow::array::StructArray::from(record_batch)).unwrap();
1207        let mut rnd_row = Row::default();
1208
1209        reader.read(0, &mut rnd_row).unwrap();
1210        assert_eq!(&og_row, &rnd_row);
1211
1212        // Create a packer to clear the row alloc.
1213        rnd_row.packer();
1214
1215        reader.read(1, &mut rnd_row).unwrap();
1216        assert_eq!(&null_row, &rnd_row);
1217    }
1218
1219    /// Regression: an array column must survive a builder -> reader round-trip.
1220    /// The builder encodes an array as a struct of `{items, dimensions}`; the
1221    /// reader must reverse that back into a `Datum::Array`.
1222    #[mz_ore::test]
1223    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
1224    fn smoketest_array() {
1225        let desc = RelationDesc::builder()
1226            .with_column(
1227                "arr",
1228                SqlScalarType::Array(Box::new(SqlScalarType::Int32)).nullable(true),
1229            )
1230            .finish();
1231
1232        let mut row_1d = Row::default();
1233        row_1d
1234            .packer()
1235            .try_push_array(
1236                &[ArrayDimension {
1237                    lower_bound: 1,
1238                    length: 3,
1239                }],
1240                [Datum::Int32(1), Datum::Null, Datum::Int32(3)],
1241            )
1242            .unwrap();
1243
1244        let mut row_empty = Row::default();
1245        row_empty
1246            .packer()
1247            .try_push_array(&[], std::iter::empty::<Datum>())
1248            .unwrap();
1249
1250        let row_null = Row::pack(vec![Datum::Null]);
1251
1252        // Encode with the builder, decode with the reader.
1253        let mut builder = crate::builder::ArrowBuilder::new(&desc, 3, 128).unwrap();
1254        builder.add_row(&row_1d).unwrap();
1255        builder.add_row(&row_empty).unwrap();
1256        builder.add_row(&row_null).unwrap();
1257        let record_batch = builder.to_record_batch().unwrap();
1258
1259        let reader = ArrowReader::new(&desc, StructArray::from(record_batch)).unwrap();
1260        let mut got = Row::default();
1261
1262        reader.read(0, &mut got).unwrap();
1263        assert_eq!(got, row_1d, "1-D array did not round-trip");
1264
1265        got.packer();
1266        reader.read(1, &mut got).unwrap();
1267        assert_eq!(got, row_empty, "empty array did not round-trip");
1268
1269        got.packer();
1270        reader.read(2, &mut got).unwrap();
1271        assert_eq!(got, row_null, "NULL array did not round-trip");
1272    }
1273
1274    #[mz_ore::test]
1275    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
1276    fn smoketest_decimal128() {
1277        let desc = RelationDesc::builder()
1278            .with_column(
1279                "a",
1280                SqlScalarType::Numeric { max_scale: None }.nullable(true),
1281            )
1282            .finish();
1283
1284        let mut dec128 = arrow::array::Decimal128Builder::new();
1285        dec128 = dec128.with_precision_and_scale(12, 3).unwrap();
1286
1287        // 1.234
1288        dec128.append_value(1234);
1289        dec128.append_null();
1290        // 100000000.009
1291        dec128.append_value(100000000009);
1292
1293        let dec128 = dec128.finish();
1294        #[allow(clippy::as_conversions)]
1295        let batch = StructArray::from(vec![(
1296            Arc::new(Field::new("a", dec128.data_type().clone(), true)),
1297            Arc::new(dec128) as arrow::array::ArrayRef,
1298        )]);
1299
1300        // Decode our data!
1301        let reader = ArrowReader::new(&desc, batch).unwrap();
1302        let mut rnd_row = Row::default();
1303
1304        reader.read(0, &mut rnd_row).unwrap();
1305        let num = rnd_row.into_element().unwrap_numeric();
1306        assert_eq!(num.0, Numeric::from(1.234f64));
1307
1308        // Create a packer to clear the row alloc.
1309        rnd_row.packer();
1310
1311        reader.read(1, &mut rnd_row).unwrap();
1312        let num = rnd_row.into_element();
1313        assert_eq!(num, Datum::Null);
1314
1315        // Create a packer to clear the row alloc.
1316        rnd_row.packer();
1317
1318        reader.read(2, &mut rnd_row).unwrap();
1319        let num = rnd_row.into_element().unwrap_numeric();
1320        assert_eq!(num.0, Numeric::from(100000000.009f64));
1321    }
1322
1323    #[mz_ore::test]
1324    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
1325    fn smoketest_decimal256() {
1326        let desc = RelationDesc::builder()
1327            .with_column(
1328                "a",
1329                SqlScalarType::Numeric { max_scale: None }.nullable(true),
1330            )
1331            .finish();
1332
1333        let mut dec256 = arrow::array::Decimal256Builder::new();
1334        dec256 = dec256.with_precision_and_scale(12, 3).unwrap();
1335
1336        // 1.234
1337        dec256.append_value(arrow::datatypes::i256::from(1234));
1338        dec256.append_null();
1339        // 100000000.009
1340        dec256.append_value(arrow::datatypes::i256::from(100000000009i64));
1341
1342        let dec256 = dec256.finish();
1343        #[allow(clippy::as_conversions)]
1344        let batch = StructArray::from(vec![(
1345            Arc::new(Field::new("a", dec256.data_type().clone(), true)),
1346            Arc::new(dec256) as arrow::array::ArrayRef,
1347        )]);
1348
1349        // Decode our data!
1350        let reader = ArrowReader::new(&desc, batch).unwrap();
1351        let mut rnd_row = Row::default();
1352
1353        reader.read(0, &mut rnd_row).unwrap();
1354        let num = rnd_row.into_element().unwrap_numeric();
1355        assert_eq!(num.0, Numeric::from(1.234f64));
1356
1357        // Create a packer to clear the row alloc.
1358        rnd_row.packer();
1359
1360        reader.read(1, &mut rnd_row).unwrap();
1361        let num = rnd_row.into_element();
1362        assert_eq!(num, Datum::Null);
1363
1364        // Create a packer to clear the row alloc.
1365        rnd_row.packer();
1366
1367        reader.read(2, &mut rnd_row).unwrap();
1368        let num = rnd_row.into_element().unwrap_numeric();
1369        assert_eq!(num.0, Numeric::from(100000000.009f64));
1370    }
1371
1372    /// Regression test for SS-193: when the destination column declares a
1373    /// `max_scale`, the reader should round the decoded value to that scale
1374    /// rather than preserving the source file's scale.
1375    #[mz_ore::test]
1376    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
1377    fn decimal_applies_destination_max_scale() {
1378        use mz_repr::adt::numeric::NumericMaxScale;
1379
1380        // Destination column is numeric(10, 2): scale 2.
1381        let desc = RelationDesc::builder()
1382            .with_column(
1383                "a",
1384                SqlScalarType::Numeric {
1385                    max_scale: Some(NumericMaxScale::try_from(2_i64).unwrap()),
1386                }
1387                .nullable(true),
1388            )
1389            .finish();
1390
1391        let expected = Numeric::from(10.45f64);
1392
1393        // The source files carry scale-3 values (10.447) that don't round
1394        // evenly to the destination's scale 2.
1395        let mut dec128 = arrow::array::Decimal128Builder::new();
1396        dec128 = dec128.with_precision_and_scale(12, 3).unwrap();
1397        dec128.append_value(10447);
1398        let dec128 = dec128.finish();
1399        #[allow(clippy::as_conversions)]
1400        let batch128 = StructArray::from(vec![(
1401            Arc::new(Field::new("a", dec128.data_type().clone(), true)),
1402            Arc::new(dec128) as arrow::array::ArrayRef,
1403        )]);
1404
1405        let reader = ArrowReader::new(&desc, batch128).unwrap();
1406        let mut rnd_row = Row::default();
1407        reader.read(0, &mut rnd_row).unwrap();
1408        let num = rnd_row.into_element().unwrap_numeric();
1409        assert_eq!(num.0, expected, "Decimal128 did not round to max_scale");
1410
1411        let mut dec256 = arrow::array::Decimal256Builder::new();
1412        dec256 = dec256.with_precision_and_scale(12, 3).unwrap();
1413        dec256.append_value(arrow::datatypes::i256::from(10447));
1414        let dec256 = dec256.finish();
1415        #[allow(clippy::as_conversions)]
1416        let batch256 = StructArray::from(vec![(
1417            Arc::new(Field::new("a", dec256.data_type().clone(), true)),
1418            Arc::new(dec256) as arrow::array::ArrayRef,
1419        )]);
1420
1421        let reader = ArrowReader::new(&desc, batch256).unwrap();
1422        let mut rnd_row = Row::default();
1423        reader.read(0, &mut rnd_row).unwrap();
1424        let num = rnd_row.into_element().unwrap_numeric();
1425        assert_eq!(num.0, expected, "Decimal256 did not round to max_scale");
1426    }
1427
1428    #[mz_ore::test]
1429    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
1430    fn decimal_rescale_to_higher_scale_overflows() {
1431        use mz_repr::adt::numeric::NumericMaxScale;
1432
1433        let value: i128 = 10_i128.pow(33) - 1;
1434
1435        let build_batch = || {
1436            let mut dec128 = arrow::array::Decimal128Builder::new();
1437            dec128 = dec128.with_precision_and_scale(38, 0).unwrap();
1438            dec128.append_value(value);
1439            let dec128 = dec128.finish();
1440            #[allow(clippy::as_conversions)]
1441            StructArray::from(vec![(
1442                Arc::new(Field::new("a", dec128.data_type().clone(), true)),
1443                Arc::new(dec128) as arrow::array::ArrayRef,
1444            )])
1445        };
1446
1447        let desc_with_scale = |scale: i64| {
1448            RelationDesc::builder()
1449                .with_column(
1450                    "a",
1451                    SqlScalarType::Numeric {
1452                        max_scale: Some(NumericMaxScale::try_from(scale).unwrap()),
1453                    }
1454                    .nullable(true),
1455                )
1456                .finish()
1457        };
1458
1459        // Test working case with supportable scale then broken case.
1460        let reader = ArrowReader::new(&desc_with_scale(2), build_batch()).unwrap();
1461        let mut row = Row::default();
1462        reader
1463            .read(0, &mut row)
1464            .expect("value must decode at destination scale 2");
1465        let num = row.into_element().unwrap_numeric();
1466        let mut expected = Numeric::try_from(value).unwrap();
1467        rescale(&mut expected, 2).unwrap();
1468        assert_eq!(num.0, expected, "value did not rescale to scale 2");
1469
1470        let reader = ArrowReader::new(&desc_with_scale(8), build_batch()).unwrap();
1471        let mut row = Row::default();
1472        let err = reader
1473            .read(0, &mut row)
1474            .expect_err("value must overflow at destination scale 8");
1475
1476        assert!(
1477            format!("{err:#}").contains("exceed maximum precision"),
1478            "unexpected error: {err:#}",
1479        );
1480    }
1481
1482    /// Regression test for database-issues#11330: when a Parquet file authored
1483    /// by an external engine encodes a discrete range in non-canonical form
1484    /// (e.g. `[1,10]` for `int4range`), the reader must canonicalize it to MZ's
1485    /// internal form (`[1,11)`). Otherwise rows ingested via `COPY FROM PARQUET`
1486    /// don't compare or hash equal to logically-identical rows constructed
1487    /// inside MZ.
1488    #[mz_ore::test]
1489    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
1490    fn range_canonicalizes_noncanonical_input() {
1491        use arrow::array::ArrayRef;
1492        use arrow::datatypes::DataType;
1493        use mz_repr::adt::range::{Range, RangeLowerBound, RangeUpperBound};
1494
1495        let desc = RelationDesc::builder()
1496            .with_column(
1497                "r",
1498                SqlScalarType::Range {
1499                    element_type: Box::new(SqlScalarType::Int32),
1500                }
1501                .nullable(true),
1502            )
1503            .finish();
1504
1505        // Build a range StructArray containing two non-canonical encodings:
1506        //   row 0: `[1,10]`  -> canonicalizes to `[1,11)`
1507        //   row 1: `(5,15)`  -> canonicalizes to `[6,15)`
1508        let lower = Int32Array::from(vec![Some(1), Some(5)]);
1509        let upper = Int32Array::from(vec![Some(10), Some(15)]);
1510        let lower_inclusive = BooleanArray::from(vec![true, false]);
1511        let upper_inclusive = BooleanArray::from(vec![true, false]);
1512        let empty = BooleanArray::from(vec![false, false]);
1513
1514        #[allow(clippy::as_conversions)]
1515        let range_fields: Vec<(Arc<Field>, ArrayRef)> = vec![
1516            (
1517                Arc::new(Field::new("lower", DataType::Int32, true)),
1518                Arc::new(lower) as ArrayRef,
1519            ),
1520            (
1521                Arc::new(Field::new("upper", DataType::Int32, true)),
1522                Arc::new(upper) as ArrayRef,
1523            ),
1524            (
1525                Arc::new(Field::new("lower_inclusive", DataType::Boolean, false)),
1526                Arc::new(lower_inclusive) as ArrayRef,
1527            ),
1528            (
1529                Arc::new(Field::new("upper_inclusive", DataType::Boolean, false)),
1530                Arc::new(upper_inclusive) as ArrayRef,
1531            ),
1532            (
1533                Arc::new(Field::new("empty", DataType::Boolean, false)),
1534                Arc::new(empty) as ArrayRef,
1535            ),
1536        ];
1537        let range_struct = StructArray::from(range_fields);
1538
1539        #[allow(clippy::as_conversions)]
1540        let batch = StructArray::from(vec![(
1541            Arc::new(Field::new("r", range_struct.data_type().clone(), true)),
1542            Arc::new(range_struct) as ArrayRef,
1543        )]);
1544
1545        let reader = ArrowReader::new(&desc, batch).unwrap();
1546
1547        // Row 0: `[1,10]` -> `[1,11)`
1548        let mut got = Row::default();
1549        reader.read(0, &mut got).unwrap();
1550        let mut want = Row::default();
1551        want.packer()
1552            .push_range(Range::new(Some((
1553                RangeLowerBound {
1554                    inclusive: true,
1555                    bound: Some(Datum::Int32(1)),
1556                },
1557                RangeUpperBound {
1558                    inclusive: true,
1559                    bound: Some(Datum::Int32(10)),
1560                },
1561            ))))
1562            .unwrap();
1563        assert_eq!(got, want, "row 0: [1,10] should canonicalize to [1,11)");
1564
1565        // Row 1: `(5,15)` -> `[6,15)`
1566        let mut got = Row::default();
1567        reader.read(1, &mut got).unwrap();
1568        let mut want = Row::default();
1569        want.packer()
1570            .push_range(Range::new(Some((
1571                RangeLowerBound {
1572                    inclusive: false,
1573                    bound: Some(Datum::Int32(5)),
1574                },
1575                RangeUpperBound {
1576                    inclusive: false,
1577                    bound: Some(Datum::Int32(15)),
1578                },
1579            ))))
1580            .unwrap();
1581        assert_eq!(got, want, "row 1: (5,15) should canonicalize to [6,15)");
1582    }
1583}