1use 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
43pub struct ArrowReader {
53 len: usize,
54 readers: Vec<ColReader>,
55}
56
57impl ArrowReader {
58 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 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 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 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 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 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 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
497enum ArrayDims {
501 UInt8(UInt8Array),
502 Int32(Int32Array),
503}
504
505impl ArrayDims {
506 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
517enum 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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)] 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 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 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 rnd_row.packer();
1214
1215 reader.read(1, &mut rnd_row).unwrap();
1216 assert_eq!(&null_row, &rnd_row);
1217 }
1218
1219 #[mz_ore::test]
1223 #[cfg_attr(miri, ignore)] 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 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)] 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 dec128.append_value(1234);
1289 dec128.append_null();
1290 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 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 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 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)] 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 dec256.append_value(arrow::datatypes::i256::from(1234));
1338 dec256.append_null();
1339 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 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 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 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 #[mz_ore::test]
1376 #[cfg_attr(miri, ignore)] fn decimal_applies_destination_max_scale() {
1378 use mz_repr::adt::numeric::NumericMaxScale;
1379
1380 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 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)] 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 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 #[mz_ore::test]
1489 #[cfg_attr(miri, ignore)] 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 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 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 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}