1use crate::arrow::buffer::bit_util::sign_extend_be;
23use crate::arrow::parquet_column;
24use crate::basic::{ColumnOrder, Type as PhysicalType};
25use crate::data_type::Int96;
26use crate::errors::{ParquetError, Result};
27use crate::file::metadata::RowGroupMetaData;
28use crate::file::metadata::page_index::PageIndexProvider;
29use crate::file::page_index::column_index::ColumnIndexMetaData;
30use crate::file::statistics::Statistics as ParquetStatistics;
31use crate::schema::types::SchemaDescriptor;
32use arrow_array::builder::{
33 BinaryBuilder, BinaryViewBuilder, BooleanBuilder, Date32Builder, Date64Builder,
34 Decimal32Builder, Decimal64Builder, FixedSizeBinaryBuilder, Float16Builder, Float32Builder,
35 Float64Builder, Int8Builder, Int16Builder, Int32Builder, Int64Builder, LargeBinaryBuilder,
36 LargeStringBuilder, StringBuilder, StringViewBuilder, Time32MillisecondBuilder,
37 Time32SecondBuilder, Time64MicrosecondBuilder, Time64NanosecondBuilder,
38 TimestampMicrosecondBuilder, TimestampMillisecondBuilder, TimestampNanosecondBuilder,
39 TimestampSecondBuilder, UInt8Builder, UInt16Builder, UInt32Builder, UInt64Builder,
40};
41use arrow_array::{
42 ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array, Decimal32Array, Decimal64Array,
43 Decimal128Array, Decimal256Array, Float16Array, Float32Array, Float64Array, Int8Array,
44 Int16Array, Int32Array, Int64Array, LargeBinaryArray, Time32MillisecondArray,
45 Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
46 TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt8Array,
47 UInt16Array, UInt32Array, UInt64Array, new_null_array,
48};
49use arrow_buffer::{NullBufferBuilder, i256};
50use arrow_schema::{DataType, Field, Schema, TimeUnit};
51use half::f16;
52use std::sync::Arc;
53
54pub(crate) fn from_bytes_to_i32(b: &[u8]) -> i32 {
57 i32::from_be_bytes(sign_extend_be::<4>(b))
61}
62
63pub(crate) fn from_bytes_to_i64(b: &[u8]) -> i64 {
66 i64::from_be_bytes(sign_extend_be::<8>(b))
67}
68
69pub(crate) fn from_bytes_to_i128(b: &[u8]) -> i128 {
72 i128::from_be_bytes(sign_extend_be::<16>(b))
73}
74
75pub(crate) fn from_bytes_to_i256(b: &[u8]) -> i256 {
78 i256::from_be_bytes(sign_extend_be::<32>(b))
79}
80
81pub(crate) fn from_bytes_to_f16(b: &[u8]) -> Option<f16> {
83 match b {
84 [low, high] => Some(f16::from_be_bytes([*high, *low])),
85 _ => None,
86 }
87}
88
89fn int96_statistics(min: &Int96, max: &Int96, unit: &TimeUnit) -> Option<(i64, i64)> {
92 let convert = |value: &Int96| {
93 const NANOS_PER_DAY: u64 = 86_400_000_000_000;
94 let words = value.data();
95 let nanos = (u64::from(words[1]) << 32) | u64::from(words[0]);
96 if nanos >= NANOS_PER_DAY {
97 return None;
98 }
99 let divisor = match unit {
100 TimeUnit::Second => 1_000_000_000,
101 TimeUnit::Millisecond => 1_000_000,
102 TimeUnit::Microsecond => 1_000,
103 TimeUnit::Nanosecond => 1,
104 };
105 let timestamp = i128::from(value.to_seconds()) * i128::from(1_000_000_000 / divisor)
109 + i128::from((nanos % 1_000_000_000) / divisor);
110 i64::try_from(timestamp).ok()
111 };
112 Some((convert(min)?, convert(max)?))
113}
114
115macro_rules! make_stats_iterator {
127 ($iterator_type:ident, $func:ident, $parquet_statistics_type:path, $stat_value_type:ty) => {
128 struct $iterator_type<'a, I>
135 where
136 I: Iterator<Item = Option<&'a ParquetStatistics>>,
137 {
138 iter: I,
139 }
140
141 impl<'a, I> $iterator_type<'a, I>
142 where
143 I: Iterator<Item = Option<&'a ParquetStatistics>>,
144 {
145 fn new(iter: I) -> Self {
147 Self { iter }
148 }
149 }
150
151 impl<'a, I> Iterator for $iterator_type<'a, I>
153 where
154 I: Iterator<Item = Option<&'a ParquetStatistics>>,
155 {
156 type Item = Option<&'a $stat_value_type>;
157
158 fn next(&mut self) -> Option<Self::Item> {
160 let next = self.iter.next();
161 next.map(|x| {
162 x.and_then(|stats| match stats {
163 $parquet_statistics_type(s) => s.$func(),
164 _ => None,
165 })
166 })
167 }
168
169 fn size_hint(&self) -> (usize, Option<usize>) {
170 self.iter.size_hint()
171 }
172 }
173 };
174}
175
176make_stats_iterator!(
177 MinBooleanStatsIterator,
178 min_opt,
179 ParquetStatistics::Boolean,
180 bool
181);
182make_stats_iterator!(
183 MaxBooleanStatsIterator,
184 max_opt,
185 ParquetStatistics::Boolean,
186 bool
187);
188make_stats_iterator!(
189 MinInt32StatsIterator,
190 min_opt,
191 ParquetStatistics::Int32,
192 i32
193);
194make_stats_iterator!(
195 MaxInt32StatsIterator,
196 max_opt,
197 ParquetStatistics::Int32,
198 i32
199);
200make_stats_iterator!(
201 MinInt64StatsIterator,
202 min_opt,
203 ParquetStatistics::Int64,
204 i64
205);
206make_stats_iterator!(
207 MaxInt64StatsIterator,
208 max_opt,
209 ParquetStatistics::Int64,
210 i64
211);
212make_stats_iterator!(
213 MinFloatStatsIterator,
214 min_opt,
215 ParquetStatistics::Float,
216 f32
217);
218make_stats_iterator!(
219 MaxFloatStatsIterator,
220 max_opt,
221 ParquetStatistics::Float,
222 f32
223);
224make_stats_iterator!(
225 MinDoubleStatsIterator,
226 min_opt,
227 ParquetStatistics::Double,
228 f64
229);
230make_stats_iterator!(
231 MaxDoubleStatsIterator,
232 max_opt,
233 ParquetStatistics::Double,
234 f64
235);
236make_stats_iterator!(
237 MinByteArrayStatsIterator,
238 min_bytes_opt,
239 ParquetStatistics::ByteArray,
240 [u8]
241);
242make_stats_iterator!(
243 MaxByteArrayStatsIterator,
244 max_bytes_opt,
245 ParquetStatistics::ByteArray,
246 [u8]
247);
248make_stats_iterator!(
249 MinFixedLenByteArrayStatsIterator,
250 min_bytes_opt,
251 ParquetStatistics::FixedLenByteArray,
252 [u8]
253);
254make_stats_iterator!(
255 MaxFixedLenByteArrayStatsIterator,
256 max_bytes_opt,
257 ParquetStatistics::FixedLenByteArray,
258 [u8]
259);
260
261macro_rules! make_decimal_stats_iterator {
279 ($iterator_type:ident, $func:ident, $bytes_func:ident, $stat_value_type:ident, $convert_func: ident) => {
280 struct $iterator_type<'a, I>
281 where
282 I: Iterator<Item = Option<&'a ParquetStatistics>>,
283 {
284 iter: I,
285 }
286
287 impl<'a, I> $iterator_type<'a, I>
288 where
289 I: Iterator<Item = Option<&'a ParquetStatistics>>,
290 {
291 fn new(iter: I) -> Self {
292 Self { iter }
293 }
294 }
295
296 impl<'a, I> Iterator for $iterator_type<'a, I>
297 where
298 I: Iterator<Item = Option<&'a ParquetStatistics>>,
299 {
300 type Item = Option<$stat_value_type>;
301
302 fn next(&mut self) -> Option<Self::Item> {
303 let next = self.iter.next();
304 next.map(|x| {
305 x.and_then(|stats| match stats {
306 ParquetStatistics::Int32(s) => {
307 s.$func().map(|x| $stat_value_type::from(*x))
308 }
309 ParquetStatistics::Int64(s) => s
310 .$func()
311 .map(|x| $stat_value_type::try_from(*x).ok())
312 .flatten(),
313 ParquetStatistics::ByteArray(s) => s.$bytes_func().map($convert_func),
314 ParquetStatistics::FixedLenByteArray(s) => {
315 s.$bytes_func().map($convert_func)
316 }
317 _ => None,
318 })
319 })
320 }
321
322 fn size_hint(&self) -> (usize, Option<usize>) {
323 self.iter.size_hint()
324 }
325 }
326 };
327}
328
329make_decimal_stats_iterator!(
330 MinDecimal32StatsIterator,
331 min_opt,
332 min_bytes_opt,
333 i32,
334 from_bytes_to_i32
335);
336make_decimal_stats_iterator!(
337 MaxDecimal32StatsIterator,
338 max_opt,
339 max_bytes_opt,
340 i32,
341 from_bytes_to_i32
342);
343make_decimal_stats_iterator!(
344 MinDecimal64StatsIterator,
345 min_opt,
346 min_bytes_opt,
347 i64,
348 from_bytes_to_i64
349);
350make_decimal_stats_iterator!(
351 MaxDecimal64StatsIterator,
352 max_opt,
353 max_bytes_opt,
354 i64,
355 from_bytes_to_i64
356);
357make_decimal_stats_iterator!(
358 MinDecimal128StatsIterator,
359 min_opt,
360 min_bytes_opt,
361 i128,
362 from_bytes_to_i128
363);
364make_decimal_stats_iterator!(
365 MaxDecimal128StatsIterator,
366 max_opt,
367 max_bytes_opt,
368 i128,
369 from_bytes_to_i128
370);
371make_decimal_stats_iterator!(
372 MinDecimal256StatsIterator,
373 min_opt,
374 min_bytes_opt,
375 i256,
376 from_bytes_to_i256
377);
378make_decimal_stats_iterator!(
379 MaxDecimal256StatsIterator,
380 max_opt,
381 max_bytes_opt,
382 i256,
383 from_bytes_to_i256
384);
385
386macro_rules! make_timestamp_stats_iterator {
388 ($iterator_type:ident, $func:ident, $bound:tt) => {
389 struct $iterator_type<'a, I>
390 where
391 I: Iterator<Item = Option<&'a ParquetStatistics>>,
392 {
393 iter: I,
394 unit: TimeUnit,
395 }
396
397 impl<'a, I> $iterator_type<'a, I>
398 where
399 I: Iterator<Item = Option<&'a ParquetStatistics>>,
400 {
401 fn new(iter: I, unit: TimeUnit) -> Self {
402 Self { iter, unit }
403 }
404 }
405
406 impl<'a, I> Iterator for $iterator_type<'a, I>
407 where
408 I: Iterator<Item = Option<&'a ParquetStatistics>>,
409 {
410 type Item = Option<i64>;
411
412 fn next(&mut self) -> Option<Self::Item> {
413 self.iter.next().map(|statistics| {
414 let statistics = statistics?;
415 match statistics {
416 ParquetStatistics::Int64(values) => values.$func().copied(),
417 ParquetStatistics::Int96(values) if !statistics.is_min_max_deprecated() => {
420 int96_statistics(values.min_opt()?, values.max_opt()?, &self.unit)
421 .map(|bounds| bounds.$bound)
422 }
423 _ => None,
424 }
425 })
426 }
427
428 fn size_hint(&self) -> (usize, Option<usize>) {
429 self.iter.size_hint()
430 }
431 }
432 };
433}
434
435make_timestamp_stats_iterator!(MinTimestampStatsIterator, min_opt, 0);
436make_timestamp_stats_iterator!(MaxTimestampStatsIterator, max_opt, 1);
437
438macro_rules! get_statistics {
445 (Min, $data_type: ident, $iterator: ident, $physical_type: ident) => {
446 get_statistics!(
447 $data_type,
448 $iterator,
449 $physical_type,
450 MinBooleanStatsIterator,
451 MinInt32StatsIterator,
452 MinInt64StatsIterator,
453 MinTimestampStatsIterator,
454 MinFloatStatsIterator,
455 MinDoubleStatsIterator,
456 MinByteArrayStatsIterator,
457 MinFixedLenByteArrayStatsIterator,
458 MinDecimal32StatsIterator,
459 MinDecimal64StatsIterator,
460 MinDecimal128StatsIterator,
461 MinDecimal256StatsIterator,
462 min_statistics
463 )
464 };
465 (Max, $data_type: ident, $iterator: ident, $physical_type: ident) => {
466 get_statistics!(
467 $data_type,
468 $iterator,
469 $physical_type,
470 MaxBooleanStatsIterator,
471 MaxInt32StatsIterator,
472 MaxInt64StatsIterator,
473 MaxTimestampStatsIterator,
474 MaxFloatStatsIterator,
475 MaxDoubleStatsIterator,
476 MaxByteArrayStatsIterator,
477 MaxFixedLenByteArrayStatsIterator,
478 MaxDecimal32StatsIterator,
479 MaxDecimal64StatsIterator,
480 MaxDecimal128StatsIterator,
481 MaxDecimal256StatsIterator,
482 max_statistics
483 )
484 };
485 (
486 $data_type: ident,
487 $iterator: ident,
488 $physical_type: ident,
489 $boolean_iter: ident,
490 $int32_iter: ident,
491 $int64_iter: ident,
492 $timestamp_iter: ident,
493 $float_iter: ident,
494 $double_iter: ident,
495 $byte_array_iter: ident,
496 $fixed_len_byte_array_iter: ident,
497 $decimal32_iter: ident,
498 $decimal64_iter: ident,
499 $decimal128_iter: ident,
500 $decimal256_iter: ident,
501 $dictionary_statistics: ident
502 ) => {
503 match $data_type {
504 DataType::Boolean => Ok(Arc::new(BooleanArray::from_iter(
505 $boolean_iter::new($iterator).map(|x| x.copied()),
506 ))),
507 DataType::Int8 => Ok(Arc::new(Int8Array::from_iter(
508 $int32_iter::new($iterator).map(|x| {
509 x.and_then(|x| i8::try_from(*x).ok())
510 }),
511 ))),
512 DataType::Int16 => Ok(Arc::new(Int16Array::from_iter(
513 $int32_iter::new($iterator).map(|x| {
514 x.and_then(|x| i16::try_from(*x).ok())
515 }),
516 ))),
517 DataType::Int32 => Ok(Arc::new(Int32Array::from_iter(
518 $int32_iter::new($iterator).map(|x| x.copied()),
519 ))),
520 DataType::Int64 => Ok(Arc::new(Int64Array::from_iter(
521 $int64_iter::new($iterator).map(|x| x.copied()),
522 ))),
523 DataType::UInt8 => Ok(Arc::new(UInt8Array::from_iter(
524 $int32_iter::new($iterator).map(|x| {
525 x.and_then(|x| u8::try_from(*x).ok())
526 }),
527 ))),
528 DataType::UInt16 => Ok(Arc::new(UInt16Array::from_iter(
529 $int32_iter::new($iterator).map(|x| {
530 x.and_then(|x| u16::try_from(*x).ok())
531 }),
532 ))),
533 DataType::UInt32 => Ok(Arc::new(UInt32Array::from_iter(
534 $int32_iter::new($iterator).map(|x| x.map(|x| *x as u32)),
535 ))),
536 DataType::UInt64 => Ok(Arc::new(UInt64Array::from_iter(
537 $int64_iter::new($iterator).map(|x| x.map(|x| *x as u64)),
538 ))),
539 DataType::Float16 => Ok(Arc::new(Float16Array::from_iter(
540 $fixed_len_byte_array_iter::new($iterator).map(|x| x.and_then(|x| {
541 from_bytes_to_f16(x)
542 })),
543 ))),
544 DataType::Float32 => Ok(Arc::new(Float32Array::from_iter(
545 $float_iter::new($iterator).map(|x| x.copied()),
546 ))),
547 DataType::Float64 => Ok(Arc::new(Float64Array::from_iter(
548 $double_iter::new($iterator).map(|x| x.copied()),
549 ))),
550 DataType::Date32 => Ok(Arc::new(Date32Array::from_iter(
551 $int32_iter::new($iterator).map(|x| x.copied()),
552 ))),
553 DataType::Date64 if $physical_type == Some(PhysicalType::INT32) => Ok(Arc::new(Date64Array::from_iter(
554 $int32_iter::new($iterator)
555 .map(|x| x.map(|x| i64::from(*x) * 24 * 60 * 60 * 1000))))),
556 DataType::Date64 if $physical_type == Some(PhysicalType::INT64) => Ok(Arc::new(Date64Array::from_iter(
557 $int64_iter::new($iterator).map(|x| x.copied()),))),
558 DataType::Timestamp(unit, timezone) =>{
559 let iter = $timestamp_iter::new($iterator, *unit);
560 Ok(match unit {
561 TimeUnit::Second => Arc::new(TimestampSecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
562 TimeUnit::Millisecond => Arc::new(TimestampMillisecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
563 TimeUnit::Microsecond => Arc::new(TimestampMicrosecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
564 TimeUnit::Nanosecond => Arc::new(TimestampNanosecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
565 })
566 },
567 DataType::Time32(unit) => {
568 Ok(match unit {
569 TimeUnit::Second => Arc::new(Time32SecondArray::from_iter(
570 $int32_iter::new($iterator).map(|x| x.copied()),
571 )),
572 TimeUnit::Millisecond => Arc::new(Time32MillisecondArray::from_iter(
573 $int32_iter::new($iterator).map(|x| x.copied()),
574 )),
575 _ => {
576 let len = $iterator.count();
577 new_null_array($data_type, len)
579 }
580 })
581 },
582 DataType::Time64(unit) => {
583 Ok(match unit {
584 TimeUnit::Microsecond => Arc::new(Time64MicrosecondArray::from_iter(
585 $int64_iter::new($iterator).map(|x| x.copied()),
586 )),
587 TimeUnit::Nanosecond => Arc::new(Time64NanosecondArray::from_iter(
588 $int64_iter::new($iterator).map(|x| x.copied()),
589 )),
590 _ => {
591 let len = $iterator.count();
592 new_null_array($data_type, len)
594 }
595 })
596 },
597 DataType::Binary => Ok(Arc::new(BinaryArray::from_iter(
598 $byte_array_iter::new($iterator)
599 ))),
600 DataType::LargeBinary => Ok(Arc::new(LargeBinaryArray::from_iter(
601 $byte_array_iter::new($iterator)
602 ))),
603 DataType::Utf8 => {
604 let iterator = $byte_array_iter::new($iterator);
605 let mut builder = StringBuilder::new();
606 for x in iterator {
607 let Some(x) = x else {
608 builder.append_null(); continue;
610 };
611
612 let Ok(x) = std::str::from_utf8(x) else {
613 builder.append_null();
614 continue;
615 };
616
617 builder.append_value(x);
618 }
619 Ok(Arc::new(builder.finish()))
620 },
621 DataType::LargeUtf8 => {
622 let iterator = $byte_array_iter::new($iterator);
623 let mut builder = LargeStringBuilder::new();
624 for x in iterator {
625 let Some(x) = x else {
626 builder.append_null(); continue;
628 };
629
630 let Ok(x) = std::str::from_utf8(x) else {
631 builder.append_null();
632 continue;
633 };
634
635 builder.append_value(x);
636 }
637 Ok(Arc::new(builder.finish()))
638 },
639 DataType::FixedSizeBinary(size) => {
640 let iterator = $fixed_len_byte_array_iter::new($iterator);
641 let mut builder = FixedSizeBinaryBuilder::new(*size);
642 for x in iterator {
643 let Some(x) = x else {
644 builder.append_null(); continue;
646 };
647
648 if x.len().try_into() != Ok(*size){
650 builder.append_null();
651 continue;
652 }
653
654 builder.append_value(x).expect("ensure to append successfully here, because size have been checked before");
655 }
656 Ok(Arc::new(builder.finish()))
657 },
658 DataType::Decimal32(precision, scale) => {
659 let arr = Decimal32Array::from_iter(
660 $decimal32_iter::new($iterator)
661 ).with_precision_and_scale(*precision, *scale)?;
662 Ok(Arc::new(arr))
663 },
664 DataType::Decimal64(precision, scale) => {
665 let arr = Decimal64Array::from_iter(
666 $decimal64_iter::new($iterator)
667 ).with_precision_and_scale(*precision, *scale)?;
668 Ok(Arc::new(arr))
669 },
670 DataType::Decimal128(precision, scale) => {
671 let arr = Decimal128Array::from_iter(
672 $decimal128_iter::new($iterator)
673 ).with_precision_and_scale(*precision, *scale)?;
674 Ok(Arc::new(arr))
675 },
676 DataType::Decimal256(precision, scale) => {
677 let arr = Decimal256Array::from_iter(
678 $decimal256_iter::new($iterator)
679 ).with_precision_and_scale(*precision, *scale)?;
680 Ok(Arc::new(arr))
681 },
682 DataType::Dictionary(_, value_type) => {
683 $dictionary_statistics(value_type, $iterator, $physical_type)
684 },
685 DataType::Utf8View => {
686 let iterator = $byte_array_iter::new($iterator);
687 let mut builder = StringViewBuilder::new();
688 for x in iterator {
689 let Some(x) = x else {
690 builder.append_null(); continue;
692 };
693
694 let Ok(x) = std::str::from_utf8(x) else {
695 builder.append_null();
696 continue;
697 };
698
699 builder.append_value(x);
700 }
701 Ok(Arc::new(builder.finish()))
702 },
703 DataType::BinaryView => {
704 let iterator = $byte_array_iter::new($iterator);
705 let mut builder = BinaryViewBuilder::new();
706 for x in iterator {
707 let Some(x) = x else {
708 builder.append_null(); continue;
710 };
711
712 builder.append_value(x);
713 }
714 Ok(Arc::new(builder.finish()))
715 }
716
717 DataType::Map(_,_) |
718 DataType::Duration(_) |
719 DataType::Interval(_) |
720 DataType::Date64 | DataType::Null |
722 DataType::List(_) |
723 DataType::ListView(_) |
724 DataType::FixedSizeList(_, _) |
725 DataType::LargeList(_) |
726 DataType::LargeListView(_) |
727 DataType::Struct(_) |
728 DataType::Union(_, _) |
729 DataType::RunEndEncoded(_, _) => {
730 let len = $iterator.count();
731 Ok(new_null_array($data_type, len))
733 }
734 }
735 };
736}
737
738macro_rules! get_data_page_statistics {
739 (Min, $data_type: ident, $iterator: ident, $physical_type: ident) => {
740 get_data_page_statistics!(
741 $data_type,
742 $iterator,
743 $physical_type,
744 min_values_iter,
745 true,
746 min_page_statistics
747 )
748 };
749 (Max, $data_type: ident, $iterator: ident, $physical_type: ident) => {
750 get_data_page_statistics!(
751 $data_type,
752 $iterator,
753 $physical_type,
754 max_values_iter,
755 false,
756 max_page_statistics
757 )
758 };
759 (
760 $data_type: ident,
761 $iterator: ident,
762 $physical_type: ident,
763 $values_iter: ident,
764 $is_min: expr,
765 $page_statistics: ident
766 ) => {{
767 let chunks: Vec<(usize, Option<&ColumnIndexMetaData>)> = $iterator.collect();
768 let capacity: usize = chunks.iter().map(|c| c.0).sum();
769 match $data_type {
770 DataType::Boolean => {
771 let mut b = BooleanBuilder::with_capacity(capacity);
772 for (len, index) in chunks {
773 match index {
774 Some(ColumnIndexMetaData::BOOLEAN(index)) => {
775 for val in index.$values_iter() {
776 b.append_option(val.copied());
777 }
778 }
779 _ => b.append_nulls(len),
780 }
781 }
782 Ok(Arc::new(b.finish()))
783 },
784 DataType::UInt8 => {
785 let mut b = UInt8Builder::with_capacity(capacity);
786 for (len, index) in chunks {
787 match index {
788 Some(ColumnIndexMetaData::INT32(index)) => {
789 b.extend_from_iter_option(
790 index.$values_iter()
791 .map(|val| val.and_then(|&x| u8::try_from(x).ok())),
792 );
793 }
794 _ => b.append_nulls(len),
795 }
796 }
797 Ok(Arc::new(b.finish()))
798 },
799 DataType::UInt16 => {
800 let mut b = UInt16Builder::with_capacity(capacity);
801 for (len, index) in chunks {
802 match index {
803 Some(ColumnIndexMetaData::INT32(index)) => {
804 b.extend_from_iter_option(
805 index.$values_iter()
806 .map(|val| val.and_then(|&x| u16::try_from(x).ok())),
807 );
808 }
809 _ => b.append_nulls(len),
810 }
811 }
812 Ok(Arc::new(b.finish()))
813 },
814 DataType::UInt32 => {
815 let mut b = UInt32Builder::with_capacity(capacity);
816 for (len, index) in chunks {
817 match index {
818 Some(ColumnIndexMetaData::INT32(index)) => {
819 b.extend_from_iter_option(
820 index.$values_iter()
821 .map(|val| val.map(|&x| x as u32)),
822 );
823 }
824 _ => b.append_nulls(len),
825 }
826 }
827 Ok(Arc::new(b.finish()))
828 },
829 DataType::UInt64 => {
830 let mut b = UInt64Builder::with_capacity(capacity);
831 for (len, index) in chunks {
832 match index {
833 Some(ColumnIndexMetaData::INT64(index)) => {
834 b.extend_from_iter_option(
835 index.$values_iter()
836 .map(|val| val.map(|&x| x as u64)),
837 );
838 }
839 _ => b.append_nulls(len),
840 }
841 }
842 Ok(Arc::new(b.finish()))
843 },
844 DataType::Int8 => {
845 let mut b = Int8Builder::with_capacity(capacity);
846 for (len, index) in chunks {
847 match index {
848 Some(ColumnIndexMetaData::INT32(index)) => {
849 b.extend_from_iter_option(
850 index.$values_iter()
851 .map(|val| val.and_then(|&x| i8::try_from(x).ok())),
852 );
853 }
854 _ => b.append_nulls(len),
855 }
856 }
857 Ok(Arc::new(b.finish()))
858 },
859 DataType::Int16 => {
860 let mut b = Int16Builder::with_capacity(capacity);
861 for (len, index) in chunks {
862 match index {
863 Some(ColumnIndexMetaData::INT32(index)) => {
864 b.extend_from_iter_option(
865 index.$values_iter()
866 .map(|val| val.and_then(|&x| i16::try_from(x).ok())),
867 );
868 }
869 _ => b.append_nulls(len),
870 }
871 }
872 Ok(Arc::new(b.finish()))
873 },
874 DataType::Int32 => {
875 let mut b = Int32Builder::with_capacity(capacity);
876 for (len, index) in chunks {
877 match index {
878 Some(ColumnIndexMetaData::INT32(index)) => {
879 b.extend_from_iter_option(
880 index.$values_iter()
881 .map(|val| val.copied()),
882 );
883 }
884 _ => b.append_nulls(len),
885 }
886 }
887 Ok(Arc::new(b.finish()))
888 },
889 DataType::Int64 => {
890 let mut b = Int64Builder::with_capacity(capacity);
891 for (len, index) in chunks {
892 match index {
893 Some(ColumnIndexMetaData::INT64(index)) => {
894 b.extend_from_iter_option(
895 index.$values_iter()
896 .map(|val| val.copied()),
897 );
898 }
899 _ => b.append_nulls(len),
900 }
901 }
902 Ok(Arc::new(b.finish()))
903 },
904 DataType::Float16 => {
905 let mut b = Float16Builder::with_capacity(capacity);
906 for (len, index) in chunks {
907 match index {
908 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
909 b.extend_from_iter_option(
910 index.$values_iter()
911 .map(|val| val.and_then(|x| from_bytes_to_f16(x))),
912 );
913 }
914 _ => b.append_nulls(len),
915 }
916 }
917 Ok(Arc::new(b.finish()))
918 },
919 DataType::Float32 => {
920 let mut b = Float32Builder::with_capacity(capacity);
921 for (len, index) in chunks {
922 match index {
923 Some(ColumnIndexMetaData::FLOAT(index)) => {
924 b.extend_from_iter_option(
925 index.$values_iter()
926 .map(|val| val.copied()),
927 );
928 }
929 _ => b.append_nulls(len),
930 }
931 }
932 Ok(Arc::new(b.finish()))
933 },
934 DataType::Float64 => {
935 let mut b = Float64Builder::with_capacity(capacity);
936 for (len, index) in chunks {
937 match index {
938 Some(ColumnIndexMetaData::DOUBLE(index)) => {
939 b.extend_from_iter_option(
940 index.$values_iter()
941 .map(|val| val.copied()),
942 );
943 }
944 _ => b.append_nulls(len),
945 }
946 }
947 Ok(Arc::new(b.finish()))
948 },
949 DataType::Binary => {
950 let mut b = BinaryBuilder::with_capacity(capacity, capacity * 10);
951 for (len, index) in chunks {
952 match index {
953 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
954 for val in index.$values_iter() {
955 b.append_option(val.map(|x| x.as_ref()));
956 }
957 }
958 _ => b.append_nulls(len),
959 }
960 }
961 Ok(Arc::new(b.finish()))
962 },
963 DataType::LargeBinary => {
964 let mut b = LargeBinaryBuilder::with_capacity(capacity, capacity * 10);
965 for (len, index) in chunks {
966 match index {
967 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
968 for val in index.$values_iter() {
969 b.append_option(val.map(|x| x.as_ref()));
970 }
971 }
972 _ => b.append_nulls(len),
973 }
974 }
975 Ok(Arc::new(b.finish()))
976 },
977 DataType::Utf8 => {
978 let mut b = StringBuilder::with_capacity(capacity, capacity * 10);
979 for (len, index) in chunks {
980 match index {
981 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
982 for val in index.$values_iter() {
983 match val {
984 Some(x) => match std::str::from_utf8(x.as_ref()) {
985 Ok(s) => b.append_value(s),
986 _ => b.append_null(),
987 }
988 None => b.append_null(),
989 }
990 }
991 }
992 _ => b.append_nulls(len),
993 }
994 }
995 Ok(Arc::new(b.finish()))
996 },
997 DataType::LargeUtf8 => {
998 let mut b = LargeStringBuilder::with_capacity(capacity, capacity * 10);
999 for (len, index) in chunks {
1000 match index {
1001 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1002 for val in index.$values_iter() {
1003 match val {
1004 Some(x) => match std::str::from_utf8(x.as_ref()) {
1005 Ok(s) => b.append_value(s),
1006 _ => b.append_null(),
1007 }
1008 None => b.append_null(),
1009 }
1010 }
1011 }
1012 _ => b.append_nulls(len),
1013 }
1014 }
1015 Ok(Arc::new(b.finish()))
1016 },
1017 DataType::Dictionary(_, value_type) => {
1018 $page_statistics(value_type, chunks.into_iter(), $physical_type)
1019 },
1020 DataType::Timestamp(unit, timezone) => {
1021 match unit {
1022 TimeUnit::Second => {
1023 let mut b = TimestampSecondBuilder::with_capacity(capacity);
1024 for (len, index) in chunks {
1025 match index {
1026 Some(ColumnIndexMetaData::INT64(index)) => {
1027 b.extend_from_iter_option(
1028 index.$values_iter()
1029 .map(|val| val.copied()),
1030 );
1031 }
1032 Some(ColumnIndexMetaData::INT96(index)) => {
1033 b.extend_from_iter_option(
1034 index.min_values_iter().zip(index.max_values_iter())
1035 .map(|(min, max)| {
1036 let (min, max) = int96_statistics(min?, max?, unit)?;
1037 Some(if $is_min { min } else { max })
1038 }),
1039 );
1040 }
1041 _ => b.append_nulls(len),
1042 }
1043 }
1044 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
1045 }
1046 TimeUnit::Millisecond => {
1047 let mut b = TimestampMillisecondBuilder::with_capacity(capacity);
1048 for (len, index) in chunks {
1049 match index {
1050 Some(ColumnIndexMetaData::INT64(index)) => {
1051 b.extend_from_iter_option(
1052 index.$values_iter()
1053 .map(|val| val.copied()),
1054 );
1055 }
1056 Some(ColumnIndexMetaData::INT96(index)) => {
1057 b.extend_from_iter_option(
1058 index.min_values_iter().zip(index.max_values_iter())
1059 .map(|(min, max)| {
1060 let (min, max) = int96_statistics(min?, max?, unit)?;
1061 Some(if $is_min { min } else { max })
1062 }),
1063 );
1064 }
1065 _ => b.append_nulls(len),
1066 }
1067 }
1068 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
1069 }
1070 TimeUnit::Microsecond => {
1071 let mut b = TimestampMicrosecondBuilder::with_capacity(capacity);
1072 for (len, index) in chunks {
1073 match index {
1074 Some(ColumnIndexMetaData::INT64(index)) => {
1075 b.extend_from_iter_option(
1076 index.$values_iter()
1077 .map(|val| val.copied()),
1078 );
1079 }
1080 Some(ColumnIndexMetaData::INT96(index)) => {
1081 b.extend_from_iter_option(
1082 index.min_values_iter().zip(index.max_values_iter())
1083 .map(|(min, max)| {
1084 let (min, max) = int96_statistics(min?, max?, unit)?;
1085 Some(if $is_min { min } else { max })
1086 }),
1087 );
1088 }
1089 _ => b.append_nulls(len),
1090 }
1091 }
1092 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
1093 }
1094 TimeUnit::Nanosecond => {
1095 let mut b = TimestampNanosecondBuilder::with_capacity(capacity);
1096 for (len, index) in chunks {
1097 match index {
1098 Some(ColumnIndexMetaData::INT64(index)) => {
1099 b.extend_from_iter_option(
1100 index.$values_iter()
1101 .map(|val| val.copied()),
1102 );
1103 }
1104 Some(ColumnIndexMetaData::INT96(index)) => {
1105 b.extend_from_iter_option(
1106 index.min_values_iter().zip(index.max_values_iter())
1107 .map(|(min, max)| {
1108 let (min, max) = int96_statistics(min?, max?, unit)?;
1109 Some(if $is_min { min } else { max })
1110 }),
1111 );
1112 }
1113 _ => b.append_nulls(len),
1114 }
1115 }
1116 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
1117 }
1118 }
1119 },
1120 DataType::Date32 => {
1121 let mut b = Date32Builder::with_capacity(capacity);
1122 for (len, index) in chunks {
1123 match index {
1124 Some(ColumnIndexMetaData::INT32(index)) => {
1125 b.extend_from_iter_option(
1126 index.$values_iter()
1127 .map(|val| val.copied()),
1128 );
1129 }
1130 _ => b.append_nulls(len),
1131 }
1132 }
1133 Ok(Arc::new(b.finish()))
1134 },
1135 DataType::Date64 if $physical_type == Some(PhysicalType::INT32)=> {
1136 let mut b = Date64Builder::with_capacity(capacity);
1137 for (len, index) in chunks {
1138 match index {
1139 Some(ColumnIndexMetaData::INT32(index)) => {
1140 b.extend_from_iter_option(
1141 index.$values_iter()
1142 .map(|val| val.map(|&x| (x as i64) * 24 * 60 * 60 * 1000)),
1143 );
1144 }
1145 _ => b.append_nulls(len),
1146 }
1147 }
1148 Ok(Arc::new(b.finish()))
1149 },
1150 DataType::Date64 if $physical_type == Some(PhysicalType::INT64) => {
1151 let mut b = Date64Builder::with_capacity(capacity);
1152 for (len, index) in chunks {
1153 match index {
1154 Some(ColumnIndexMetaData::INT64(index)) => {
1155 b.extend_from_iter_option(
1156 index.$values_iter()
1157 .map(|val| val.copied()),
1158 );
1159 }
1160 _ => b.append_nulls(len),
1161 }
1162 }
1163 Ok(Arc::new(b.finish()))
1164 },
1165 DataType::Decimal32(precision, scale) => {
1166 let mut b = Decimal32Builder::with_capacity(capacity);
1167 for (len, index) in chunks {
1168 match index {
1169 Some(ColumnIndexMetaData::INT32(index)) => {
1170 b.extend_from_iter_option(
1171 index.$values_iter()
1172 .map(|val| val.copied()),
1173 );
1174 }
1175 Some(ColumnIndexMetaData::INT64(index)) => {
1176 b.extend_from_iter_option(
1177 index.$values_iter()
1178 .map(|val| val.and_then(|&x| i32::try_from(x).ok())),
1179 );
1180 }
1181 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1182 b.extend_from_iter_option(
1183 index.$values_iter()
1184 .map(|val| val.map(|x| from_bytes_to_i32(x.as_ref()))),
1185 );
1186 }
1187 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1188 b.extend_from_iter_option(
1189 index.$values_iter()
1190 .map(|val| val.map(|x| from_bytes_to_i32(x.as_ref()))),
1191 );
1192 }
1193 _ => b.append_nulls(len),
1194 }
1195 }
1196 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1197 },
1198 DataType::Decimal64(precision, scale) => {
1199 let mut b = Decimal64Builder::with_capacity(capacity);
1200 for (len, index) in chunks {
1201 match index {
1202 Some(ColumnIndexMetaData::INT32(index)) => {
1203 b.extend_from_iter_option(
1204 index.$values_iter()
1205 .map(|val| val.map(|x| *x as i64)),
1206 );
1207 }
1208 Some(ColumnIndexMetaData::INT64(index)) => {
1209 b.extend_from_iter_option(
1210 index.$values_iter()
1211 .map(|val| val.copied()),
1212 );
1213 }
1214 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1215 b.extend_from_iter_option(
1216 index.$values_iter()
1217 .map(|val| val.map(|x| from_bytes_to_i64(x.as_ref()))),
1218 );
1219 }
1220 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1221 b.extend_from_iter_option(
1222 index.$values_iter()
1223 .map(|val| val.map(|x| from_bytes_to_i64(x.as_ref()))),
1224 );
1225 }
1226 _ => b.append_nulls(len),
1227 }
1228 }
1229 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1230 },
1231 DataType::Decimal128(precision, scale) => {
1232 let mut b = Decimal128Array::builder(capacity);
1233 for (len, index) in chunks {
1234 match index {
1235 Some(ColumnIndexMetaData::INT32(index)) => {
1236 b.extend_from_iter_option(
1237 index.$values_iter()
1238 .map(|val| val.map(|x| *x as i128)),
1239 );
1240 }
1241 Some(ColumnIndexMetaData::INT64(index)) => {
1242 b.extend_from_iter_option(
1243 index.$values_iter()
1244 .map(|val| val.map(|x| *x as i128)),
1245 );
1246 }
1247 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1248 b.extend_from_iter_option(
1249 index.$values_iter()
1250 .map(|val| val.map(|x| from_bytes_to_i128(x.as_ref()))),
1251 );
1252 }
1253 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1254 b.extend_from_iter_option(
1255 index.$values_iter()
1256 .map(|val| val.map(|x| from_bytes_to_i128(x.as_ref()))),
1257 );
1258 }
1259 _ => b.append_nulls(len),
1260 }
1261 }
1262 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1263 },
1264 DataType::Decimal256(precision, scale) => {
1265 let mut b = Decimal256Array::builder(capacity);
1266 for (len, index) in chunks {
1267 match index {
1268 Some(ColumnIndexMetaData::INT32(index)) => {
1269 b.extend_from_iter_option(
1270 index.$values_iter()
1271 .map(|val| val.map(|x| i256::from_i128(*x as i128))),
1272 );
1273 }
1274 Some(ColumnIndexMetaData::INT64(index)) => {
1275 b.extend_from_iter_option(
1276 index.$values_iter()
1277 .map(|val| val.map(|x| i256::from_i128(*x as i128))),
1278 );
1279 }
1280 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1281 b.extend_from_iter_option(
1282 index.$values_iter()
1283 .map(|val| val.map(|x| from_bytes_to_i256(x.as_ref()))),
1284 );
1285 }
1286 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1287 b.extend_from_iter_option(
1288 index.$values_iter()
1289 .map(|val| val.map(|x| from_bytes_to_i256(x.as_ref()))),
1290 );
1291 }
1292 _ => b.append_nulls(len),
1293 }
1294 }
1295 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1296 },
1297 DataType::Time32(unit) => {
1298 match unit {
1299 TimeUnit::Second => {
1300 let mut b = Time32SecondBuilder::with_capacity(capacity);
1301 for (len, index) in chunks {
1302 match index {
1303 Some(ColumnIndexMetaData::INT32(index)) => {
1304 b.extend_from_iter_option(
1305 index.$values_iter()
1306 .map(|val| val.copied()),
1307 );
1308 }
1309 _ => b.append_nulls(len),
1310 }
1311 }
1312 Ok(Arc::new(b.finish()))
1313 }
1314 TimeUnit::Millisecond => {
1315 let mut b = Time32MillisecondBuilder::with_capacity(capacity);
1316 for (len, index) in chunks {
1317 match index {
1318 Some(ColumnIndexMetaData::INT32(index)) => {
1319 b.extend_from_iter_option(
1320 index.$values_iter()
1321 .map(|val| val.copied()),
1322 );
1323 }
1324 _ => b.append_nulls(len),
1325 }
1326 }
1327 Ok(Arc::new(b.finish()))
1328 }
1329 _ => {
1330 Ok(new_null_array($data_type, capacity))
1331 }
1332 }
1333 }
1334 DataType::Time64(unit) => {
1335 match unit {
1336 TimeUnit::Microsecond => {
1337 let mut b = Time64MicrosecondBuilder::with_capacity(capacity);
1338 for (len, index) in chunks {
1339 match index {
1340 Some(ColumnIndexMetaData::INT64(index)) => {
1341 b.extend_from_iter_option(
1342 index.$values_iter()
1343 .map(|val| val.copied()),
1344 );
1345 }
1346 _ => b.append_nulls(len),
1347 }
1348 }
1349 Ok(Arc::new(b.finish()))
1350 }
1351 TimeUnit::Nanosecond => {
1352 let mut b = Time64NanosecondBuilder::with_capacity(capacity);
1353 for (len, index) in chunks {
1354 match index {
1355 Some(ColumnIndexMetaData::INT64(index)) => {
1356 b.extend_from_iter_option(
1357 index.$values_iter()
1358 .map(|val| val.copied()),
1359 );
1360 }
1361 _ => b.append_nulls(len),
1362 }
1363 }
1364 Ok(Arc::new(b.finish()))
1365 }
1366 _ => {
1367 Ok(new_null_array($data_type, capacity))
1368 }
1369 }
1370 },
1371 DataType::FixedSizeBinary(size) => {
1372 let mut b = FixedSizeBinaryBuilder::with_capacity(capacity, *size);
1373 for (len, index) in chunks {
1374 match index {
1375 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1376 for val in index.$values_iter() {
1377 match val {
1378 Some(v) => {
1379 if v.len() == *size as usize {
1380 let _ = b.append_value(v.as_ref())?;
1381 } else {
1382 b.append_null();
1383 }
1384 }
1385 None => b.append_null(),
1386 }
1387 }
1388 }
1389 _ => b.append_nulls(len),
1390 }
1391 }
1392 Ok(Arc::new(b.finish()))
1393 },
1394 DataType::Utf8View => {
1395 let mut b = StringViewBuilder::with_capacity(capacity);
1396 for (len, index) in chunks {
1397 match index {
1398 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1399 for val in index.$values_iter() {
1400 match val {
1401 Some(x) => match std::str::from_utf8(x.as_ref()) {
1402 Ok(s) => b.append_value(s),
1403 _ => b.append_null(),
1404 }
1405 None => b.append_null(),
1406 }
1407 }
1408 }
1409 _ => {
1410 for _ in 0..len { b.append_null(); }
1411 }
1412 }
1413 }
1414 Ok(Arc::new(b.finish()))
1415 },
1416 DataType::BinaryView => {
1417 let mut b = BinaryViewBuilder::with_capacity(capacity);
1418 for (len, index) in chunks {
1419 match index {
1420 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1421 for val in index.$values_iter() {
1422 match val {
1423 Some(v) => b.append_value(v.as_ref()),
1424 None => b.append_null(),
1425 }
1426 }
1427 }
1428 _ => {
1429 for _ in 0..len { b.append_null(); }
1430 }
1431 }
1432 }
1433 Ok(Arc::new(b.finish()))
1434 },
1435 DataType::Date64 | DataType::Null |
1437 DataType::Duration(_) |
1438 DataType::Interval(_) |
1439 DataType::List(_) |
1440 DataType::ListView(_) |
1441 DataType::FixedSizeList(_, _) |
1442 DataType::LargeList(_) |
1443 DataType::LargeListView(_) |
1444 DataType::Struct(_) |
1445 DataType::Union(_, _) |
1446 DataType::Map(_, _) |
1447 DataType::RunEndEncoded(_, _) => {
1448 Ok(new_null_array($data_type, capacity))
1450 },
1451 }
1452 }
1453 };
1454}
1455fn min_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
1460 data_type: &DataType,
1461 iterator: I,
1462 physical_type: Option<PhysicalType>,
1463) -> Result<ArrayRef> {
1464 get_statistics!(Min, data_type, iterator, physical_type)
1465}
1466
1467fn max_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
1471 data_type: &DataType,
1472 iterator: I,
1473 physical_type: Option<PhysicalType>,
1474) -> Result<ArrayRef> {
1475 get_statistics!(Max, data_type, iterator, physical_type)
1476}
1477
1478pub(crate) fn min_page_statistics<'a, I>(
1481 data_type: &DataType,
1482 iterator: I,
1483 physical_type: Option<PhysicalType>,
1484) -> Result<ArrayRef>
1485where
1486 I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
1487{
1488 get_data_page_statistics!(Min, data_type, iterator, physical_type)
1489}
1490
1491pub(crate) fn max_page_statistics<'a, I>(
1494 data_type: &DataType,
1495 iterator: I,
1496 physical_type: Option<PhysicalType>,
1497) -> Result<ArrayRef>
1498where
1499 I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
1500{
1501 get_data_page_statistics!(Max, data_type, iterator, physical_type)
1502}
1503
1504pub(crate) fn null_counts_page_statistics<'a, I>(iterator: I) -> Result<UInt64Array>
1509where
1510 I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
1511{
1512 let chunks: Vec<_> = iterator.collect();
1513 let total_capacity: usize = chunks.iter().map(|(len, _)| *len).sum();
1514 let mut values = Vec::with_capacity(total_capacity);
1515 let mut nulls = NullBufferBuilder::new(total_capacity);
1516 for (len, index) in chunks {
1517 match index {
1518 Some(index) if index.null_counts().is_some() => {
1519 values.extend(index.null_counts().unwrap().iter().map(|&x| x as u64));
1520 nulls.append_n_non_nulls(len);
1521 }
1522 _ => {
1523 values.resize(values.len() + len, 0);
1524 nulls.append_n_nulls(len);
1525 }
1526 }
1527 }
1528 let null_buffer = nulls.build();
1529 let array = UInt64Array::new(values.into(), null_buffer);
1530 Ok(array)
1531}
1532
1533pub(crate) fn nan_counts_page_statistics<'a, I>(iterator: I) -> Result<UInt64Array>
1538where
1539 I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
1540{
1541 let chunks: Vec<_> = iterator.collect();
1542 let total_capacity: usize = chunks.iter().map(|(len, _)| *len).sum();
1543 let mut values = Vec::with_capacity(total_capacity);
1544 let mut nulls = NullBufferBuilder::new(total_capacity);
1545 for (len, index) in chunks {
1546 match index {
1547 Some(index) if index.nan_counts().is_some() => {
1548 values.extend(index.nan_counts().unwrap().iter().map(|&x| x as u64));
1549 nulls.append_n_non_nulls(len);
1550 }
1551 _ => {
1552 values.resize(values.len() + len, 0);
1553 nulls.append_n_nulls(len);
1554 }
1555 }
1556 }
1557 let null_buffer = nulls.build();
1558 let array = UInt64Array::new(values.into(), null_buffer);
1559 Ok(array)
1560}
1561
1562#[derive(Debug)]
1593pub struct StatisticsConverter<'a> {
1594 parquet_column_index: Option<usize>,
1596 arrow_field: &'a Field,
1598 missing_null_counts_as_zero: bool,
1600 physical_type: Option<PhysicalType>,
1602 column_order: Option<ColumnOrder>,
1604}
1605
1606impl<'a> StatisticsConverter<'a> {
1607 pub fn parquet_column_index(&self) -> Option<usize> {
1612 self.parquet_column_index
1613 }
1614
1615 pub fn arrow_field(&self) -> &'a Field {
1617 self.arrow_field
1618 }
1619
1620 pub fn with_missing_null_counts_as_zero(mut self, missing_null_counts_as_zero: bool) -> Self {
1633 self.missing_null_counts_as_zero = missing_null_counts_as_zero;
1634 self
1635 }
1636
1637 pub fn with_column_orders(mut self, column_orders: Option<&[ColumnOrder]>) -> Self {
1649 self.column_order = self
1650 .parquet_column_index
1651 .and_then(|index| column_orders?.get(index).copied());
1652 self
1653 }
1654
1655 pub fn row_group_row_counts<I>(&self, metadatas: I) -> Result<Option<UInt64Array>>
1686 where
1687 I: IntoIterator<Item = &'a RowGroupMetaData>,
1688 {
1689 let Some(_) = self.parquet_column_index else {
1690 return Ok(None);
1691 };
1692
1693 let mut builder = UInt64Array::builder(10);
1694 for metadata in metadatas {
1695 let row_count = metadata.num_rows();
1696 let row_count: u64 = row_count.try_into().map_err(|e| {
1697 arrow_err!(format!(
1698 "Parquet row count {row_count} too large to convert to u64: {e}"
1699 ))
1700 })?;
1701 builder.append_value(row_count);
1702 }
1703 Ok(Some(builder.finish()))
1704 }
1705
1706 pub fn try_new<'b>(
1721 column_name: &'b str,
1722 arrow_schema: &'a Schema,
1723 parquet_schema: &'a SchemaDescriptor,
1724 ) -> Result<Self> {
1725 let Some((_idx, arrow_field)) = arrow_schema.column_with_name(column_name) else {
1727 return Err(arrow_err!(format!(
1728 "Column '{}' not found in schema for statistics conversion",
1729 column_name
1730 )));
1731 };
1732
1733 let parquet_index = match parquet_column(parquet_schema, arrow_schema, column_name) {
1735 Some((parquet_idx, matched_field)) => {
1736 if matched_field.as_ref() != arrow_field {
1738 return Err(arrow_err!(format!(
1739 "Matched column '{:?}' does not match original matched column '{:?}'",
1740 matched_field, arrow_field
1741 )));
1742 }
1743 Some(parquet_idx)
1744 }
1745 None => None,
1746 };
1747
1748 Ok(Self {
1749 parquet_column_index: parquet_index,
1750 arrow_field,
1751 missing_null_counts_as_zero: true,
1752 physical_type: parquet_index.map(|idx| parquet_schema.column(idx).physical_type()),
1753 column_order: None,
1754 })
1755 }
1756
1757 pub fn from_column_index(
1779 parquet_column_index: usize,
1780 arrow_field: &'a Field,
1781 parquet_schema: &'a SchemaDescriptor,
1782 ) -> Result<Self> {
1783 if parquet_column_index >= parquet_schema.columns().len() {
1784 return Err(arrow_err!(format!(
1785 "Parquet column index {} out of bounds, column count {}",
1786 parquet_column_index,
1787 parquet_schema.columns().len()
1788 )));
1789 }
1790
1791 let physical_type = parquet_schema.column(parquet_column_index).physical_type();
1792
1793 Ok(Self {
1794 parquet_column_index: Some(parquet_column_index),
1795 arrow_field,
1796 missing_null_counts_as_zero: true,
1797 physical_type: Some(physical_type),
1798 column_order: None,
1799 })
1800 }
1801
1802 pub fn row_group_mins<I>(&self, metadatas: I) -> Result<ArrayRef>
1846 where
1847 I: IntoIterator<Item = &'a RowGroupMetaData>,
1848 {
1849 let data_type = self.arrow_field.data_type();
1850
1851 let Some(parquet_index) = self.parquet_column_index else {
1852 return Ok(self.make_null_array(data_type, metadatas));
1853 };
1854
1855 let iter = metadatas.into_iter().map(|x| {
1856 x.column(parquet_index)
1857 .statistics()
1858 .filter(|_| self.is_min_max_order_supported())
1859 });
1860 min_statistics(data_type, iter, self.physical_type)
1861 }
1862
1863 pub fn row_group_maxes<I>(&self, metadatas: I) -> Result<ArrayRef>
1867 where
1868 I: IntoIterator<Item = &'a RowGroupMetaData>,
1869 {
1870 let data_type = self.arrow_field.data_type();
1871
1872 let Some(parquet_index) = self.parquet_column_index else {
1873 return Ok(self.make_null_array(data_type, metadatas));
1874 };
1875
1876 let iter = metadatas.into_iter().map(|x| {
1877 x.column(parquet_index)
1878 .statistics()
1879 .filter(|_| self.is_min_max_order_supported())
1880 });
1881 max_statistics(data_type, iter, self.physical_type)
1882 }
1883
1884 pub fn row_group_is_max_value_exact<I>(&self, metadatas: I) -> Result<BooleanArray>
1888 where
1889 I: IntoIterator<Item = &'a RowGroupMetaData>,
1890 {
1891 let Some(parquet_index) = self.parquet_column_index else {
1892 let num_row_groups = metadatas.into_iter().count();
1893 return Ok(BooleanArray::from_iter(std::iter::repeat_n(
1894 None,
1895 num_row_groups,
1896 )));
1897 };
1898
1899 let is_max_value_exact = metadatas
1900 .into_iter()
1901 .map(|x| x.column(parquet_index).statistics())
1902 .map(|s| s.map(|s| s.max_is_exact()));
1903 Ok(BooleanArray::from_iter(is_max_value_exact))
1904 }
1905
1906 pub fn row_group_is_min_value_exact<I>(&self, metadatas: I) -> Result<BooleanArray>
1910 where
1911 I: IntoIterator<Item = &'a RowGroupMetaData>,
1912 {
1913 let Some(parquet_index) = self.parquet_column_index else {
1914 let num_row_groups = metadatas.into_iter().count();
1915 return Ok(BooleanArray::from_iter(std::iter::repeat_n(
1916 None,
1917 num_row_groups,
1918 )));
1919 };
1920
1921 let is_min_value_exact = metadatas
1922 .into_iter()
1923 .map(|x| x.column(parquet_index).statistics())
1924 .map(|s| s.map(|s| s.min_is_exact()));
1925 Ok(BooleanArray::from_iter(is_min_value_exact))
1926 }
1927
1928 pub fn row_group_null_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
1932 where
1933 I: IntoIterator<Item = &'a RowGroupMetaData>,
1934 {
1935 let Some(parquet_index) = self.parquet_column_index else {
1936 let num_row_groups = metadatas.into_iter().count();
1937 return Ok(UInt64Array::from_iter(std::iter::repeat_n(
1938 None,
1939 num_row_groups,
1940 )));
1941 };
1942
1943 let null_counts = metadatas
1944 .into_iter()
1945 .map(|x| x.column(parquet_index).statistics())
1946 .map(|s| {
1947 let s = s?;
1948 if self.missing_null_counts_as_zero {
1949 Some(s.null_count_opt().unwrap_or(0))
1950 } else {
1951 s.null_count_opt()
1952 }
1953 });
1954 Ok(UInt64Array::from_iter(null_counts))
1955 }
1956
1957 pub fn row_group_nan_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
1961 where
1962 I: IntoIterator<Item = &'a RowGroupMetaData>,
1963 {
1964 let Some(parquet_index) = self.parquet_column_index else {
1965 let num_row_groups = metadatas.into_iter().count();
1966 return Ok(UInt64Array::from_iter(std::iter::repeat_n(
1967 None,
1968 num_row_groups,
1969 )));
1970 };
1971
1972 let nan_counts = metadatas
1973 .into_iter()
1974 .map(|x| x.column(parquet_index).statistics())
1975 .map(|s| s?.nan_count_opt());
1976 Ok(UInt64Array::from_iter(nan_counts))
1977 }
1978
1979 pub fn row_group_distinct_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
1983 where
1984 I: IntoIterator<Item = &'a RowGroupMetaData>,
1985 {
1986 let Some(parquet_index) = self.parquet_column_index else {
1987 let num_row_groups = metadatas.into_iter().count();
1988 return Ok(UInt64Array::from_iter(std::iter::repeat_n(
1989 None,
1990 num_row_groups,
1991 )));
1992 };
1993
1994 let distinct_counts = metadatas
1995 .into_iter()
1996 .map(|x| x.column(parquet_index).statistics())
1997 .map(|s| s?.distinct_count_opt());
1998 Ok(UInt64Array::from_iter(distinct_counts))
1999 }
2000
2001 pub fn data_page_mins<I>(
2051 &self,
2052 page_index: &dyn PageIndexProvider,
2053 row_group_indices: I,
2054 ) -> Result<ArrayRef>
2055 where
2056 I: IntoIterator<Item = &'a usize>,
2057 {
2058 let data_type = self.arrow_field.data_type();
2059
2060 let Some(parquet_index) = self.parquet_column_index else {
2061 return Ok(self.make_null_array(data_type, row_group_indices));
2062 };
2063
2064 let iter = row_group_indices.into_iter().map(|rg_index| {
2065 let column_page_index_per_row_group_per_column = page_index
2066 .column_index(*rg_index, parquet_index)
2067 .filter(|_| self.is_min_max_order_supported());
2068 let num_data_pages = page_index
2069 .num_data_pages(*rg_index, parquet_index)
2070 .unwrap_or(0);
2071
2072 (num_data_pages, column_page_index_per_row_group_per_column)
2073 });
2074
2075 min_page_statistics(data_type, iter, self.physical_type)
2076 }
2077
2078 pub fn data_page_maxes<I>(
2082 &self,
2083 page_index: &dyn PageIndexProvider,
2084 row_group_indices: I,
2085 ) -> Result<ArrayRef>
2086 where
2087 I: IntoIterator<Item = &'a usize>,
2088 {
2089 let data_type = self.arrow_field.data_type();
2090
2091 let Some(parquet_index) = self.parquet_column_index else {
2092 return Ok(self.make_null_array(data_type, row_group_indices));
2093 };
2094
2095 let iter = row_group_indices.into_iter().map(|rg_index| {
2096 let column_page_index_per_row_group_per_column = page_index
2097 .column_index(*rg_index, parquet_index)
2098 .filter(|_| self.is_min_max_order_supported());
2099 let num_data_pages = page_index
2100 .num_data_pages(*rg_index, parquet_index)
2101 .unwrap_or(0);
2102
2103 (num_data_pages, column_page_index_per_row_group_per_column)
2104 });
2105
2106 max_page_statistics(data_type, iter, self.physical_type)
2107 }
2108
2109 pub fn data_page_null_counts<I>(
2113 &self,
2114 page_index: &dyn PageIndexProvider,
2115 row_group_indices: I,
2116 ) -> Result<UInt64Array>
2117 where
2118 I: IntoIterator<Item = &'a usize>,
2119 {
2120 let Some(parquet_index) = self.parquet_column_index else {
2121 let num_row_groups = row_group_indices.into_iter().count();
2122 return Ok(UInt64Array::new_null(num_row_groups));
2123 };
2124
2125 let iter = row_group_indices.into_iter().map(|rg_index| {
2126 let column_page_index_per_row_group_per_column =
2127 page_index.column_index(*rg_index, parquet_index);
2128 let num_data_pages = page_index
2129 .num_data_pages(*rg_index, parquet_index)
2130 .unwrap_or(0);
2131
2132 (num_data_pages, column_page_index_per_row_group_per_column)
2133 });
2134 null_counts_page_statistics(iter)
2135 }
2136
2137 pub fn data_page_nan_counts<I>(
2141 &self,
2142 page_index: &dyn PageIndexProvider,
2143 row_group_indices: I,
2144 ) -> Result<UInt64Array>
2145 where
2146 I: IntoIterator<Item = &'a usize>,
2147 {
2148 let Some(parquet_index) = self.parquet_column_index else {
2149 let num_row_groups = row_group_indices.into_iter().count();
2150 return Ok(UInt64Array::new_null(num_row_groups));
2151 };
2152
2153 let iter = row_group_indices.into_iter().map(|rg_index| {
2154 let column_page_index_per_row_group_per_column =
2155 page_index.column_index(*rg_index, parquet_index);
2156 let num_data_pages = page_index
2157 .num_data_pages(*rg_index, parquet_index)
2158 .unwrap_or(0);
2159
2160 (num_data_pages, column_page_index_per_row_group_per_column)
2161 });
2162 nan_counts_page_statistics(iter)
2163 }
2164
2165 pub fn data_page_row_counts<I>(
2183 &self,
2184 page_index: &dyn PageIndexProvider,
2185 row_group_metadatas: &'a [RowGroupMetaData],
2186 row_group_indices: I,
2187 ) -> Result<Option<UInt64Array>>
2188 where
2189 I: IntoIterator<Item = &'a usize>,
2190 {
2191 let Some(parquet_index) = self.parquet_column_index else {
2192 return Ok(None);
2196 };
2197
2198 let mut row_counts = Vec::new();
2199 let mut nulls = NullBufferBuilder::new(0);
2200 for rg_idx in row_group_indices {
2201 let Some(offset_index) = page_index.offset_index(*rg_idx, parquet_index) else {
2202 let num_pages = page_index
2206 .column_index(*rg_idx, parquet_index)
2207 .map(|index| index.num_pages() as usize)
2208 .unwrap_or(0);
2209 row_counts.resize(row_counts.len() + num_pages, 0);
2210 nulls.append_n_nulls(num_pages);
2211 continue;
2212 };
2213 let page_locations = offset_index.page_locations();
2214
2215 let row_count_per_page = page_locations
2216 .windows(2)
2217 .map(|loc| Some(loc[1].first_row_index as u64 - loc[0].first_row_index as u64));
2218
2219 let num_rows_in_row_group = &row_group_metadatas[*rg_idx].num_rows();
2221 let row_count_per_page = row_count_per_page.chain(std::iter::once(Some(
2222 *num_rows_in_row_group as u64
2223 - page_locations.last().unwrap().first_row_index as u64,
2224 )));
2225
2226 row_counts.extend(row_count_per_page.clone().map(|x| x.unwrap_or(0)));
2227 for val in row_count_per_page {
2228 if val.is_some() {
2229 nulls.append_non_null();
2230 } else {
2231 nulls.append_null();
2232 }
2233 }
2234 }
2235
2236 Ok(Some(UInt64Array::new(row_counts.into(), nulls.build())))
2237 }
2238
2239 fn is_min_max_order_supported(&self) -> bool {
2240 self.physical_type != Some(PhysicalType::INT96)
2241 || self.column_order == Some(ColumnOrder::INT96_TIMESTAMP_ORDER)
2242 }
2243
2244 fn make_null_array<I, A>(&self, data_type: &DataType, metadatas: I) -> ArrayRef
2246 where
2247 I: IntoIterator<Item = A>,
2248 {
2249 let num_row_groups = metadatas.into_iter().count();
2251 new_null_array(data_type, num_row_groups)
2252 }
2253}
2254
2255