1use crate::arrow::buffer::bit_util::sign_extend_be;
23use crate::arrow::parquet_column;
24use crate::basic::Type as PhysicalType;
25use crate::errors::{ParquetError, Result};
26use crate::file::metadata::{ParquetColumnIndex, ParquetOffsetIndex, RowGroupMetaData};
27use crate::file::page_index::column_index::ColumnIndexMetaData;
28use crate::file::statistics::Statistics as ParquetStatistics;
29use crate::schema::types::SchemaDescriptor;
30use arrow_array::builder::{
31 BinaryBuilder, BinaryViewBuilder, BooleanBuilder, Date32Builder, Date64Builder,
32 Decimal32Builder, Decimal64Builder, FixedSizeBinaryBuilder, Float16Builder, Float32Builder,
33 Float64Builder, Int8Builder, Int16Builder, Int32Builder, Int64Builder, LargeBinaryBuilder,
34 LargeStringBuilder, StringBuilder, StringViewBuilder, Time32MillisecondBuilder,
35 Time32SecondBuilder, Time64MicrosecondBuilder, Time64NanosecondBuilder,
36 TimestampMicrosecondBuilder, TimestampMillisecondBuilder, TimestampNanosecondBuilder,
37 TimestampSecondBuilder, UInt8Builder, UInt16Builder, UInt32Builder, UInt64Builder,
38};
39use arrow_array::{
40 ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array, Decimal32Array, Decimal64Array,
41 Decimal128Array, Decimal256Array, Float16Array, Float32Array, Float64Array, Int8Array,
42 Int16Array, Int32Array, Int64Array, LargeBinaryArray, Time32MillisecondArray,
43 Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
44 TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt8Array,
45 UInt16Array, UInt32Array, UInt64Array, new_null_array,
46};
47use arrow_buffer::{NullBufferBuilder, i256};
48use arrow_schema::{DataType, Field, Schema, TimeUnit};
49use half::f16;
50use std::sync::Arc;
51
52pub(crate) fn from_bytes_to_i32(b: &[u8]) -> i32 {
55 i32::from_be_bytes(sign_extend_be::<4>(b))
59}
60
61pub(crate) fn from_bytes_to_i64(b: &[u8]) -> i64 {
64 i64::from_be_bytes(sign_extend_be::<8>(b))
65}
66
67pub(crate) fn from_bytes_to_i128(b: &[u8]) -> i128 {
70 i128::from_be_bytes(sign_extend_be::<16>(b))
71}
72
73pub(crate) fn from_bytes_to_i256(b: &[u8]) -> i256 {
76 i256::from_be_bytes(sign_extend_be::<32>(b))
77}
78
79pub(crate) fn from_bytes_to_f16(b: &[u8]) -> Option<f16> {
81 match b {
82 [low, high] => Some(f16::from_be_bytes([*high, *low])),
83 _ => None,
84 }
85}
86
87macro_rules! make_stats_iterator {
99 ($iterator_type:ident, $func:ident, $parquet_statistics_type:path, $stat_value_type:ty) => {
100 struct $iterator_type<'a, I>
107 where
108 I: Iterator<Item = Option<&'a ParquetStatistics>>,
109 {
110 iter: I,
111 }
112
113 impl<'a, I> $iterator_type<'a, I>
114 where
115 I: Iterator<Item = Option<&'a ParquetStatistics>>,
116 {
117 fn new(iter: I) -> Self {
119 Self { iter }
120 }
121 }
122
123 impl<'a, I> Iterator for $iterator_type<'a, I>
125 where
126 I: Iterator<Item = Option<&'a ParquetStatistics>>,
127 {
128 type Item = Option<&'a $stat_value_type>;
129
130 fn next(&mut self) -> Option<Self::Item> {
132 let next = self.iter.next();
133 next.map(|x| {
134 x.and_then(|stats| match stats {
135 $parquet_statistics_type(s) => s.$func(),
136 _ => None,
137 })
138 })
139 }
140
141 fn size_hint(&self) -> (usize, Option<usize>) {
142 self.iter.size_hint()
143 }
144 }
145 };
146}
147
148make_stats_iterator!(
149 MinBooleanStatsIterator,
150 min_opt,
151 ParquetStatistics::Boolean,
152 bool
153);
154make_stats_iterator!(
155 MaxBooleanStatsIterator,
156 max_opt,
157 ParquetStatistics::Boolean,
158 bool
159);
160make_stats_iterator!(
161 MinInt32StatsIterator,
162 min_opt,
163 ParquetStatistics::Int32,
164 i32
165);
166make_stats_iterator!(
167 MaxInt32StatsIterator,
168 max_opt,
169 ParquetStatistics::Int32,
170 i32
171);
172make_stats_iterator!(
173 MinInt64StatsIterator,
174 min_opt,
175 ParquetStatistics::Int64,
176 i64
177);
178make_stats_iterator!(
179 MaxInt64StatsIterator,
180 max_opt,
181 ParquetStatistics::Int64,
182 i64
183);
184make_stats_iterator!(
185 MinFloatStatsIterator,
186 min_opt,
187 ParquetStatistics::Float,
188 f32
189);
190make_stats_iterator!(
191 MaxFloatStatsIterator,
192 max_opt,
193 ParquetStatistics::Float,
194 f32
195);
196make_stats_iterator!(
197 MinDoubleStatsIterator,
198 min_opt,
199 ParquetStatistics::Double,
200 f64
201);
202make_stats_iterator!(
203 MaxDoubleStatsIterator,
204 max_opt,
205 ParquetStatistics::Double,
206 f64
207);
208make_stats_iterator!(
209 MinByteArrayStatsIterator,
210 min_bytes_opt,
211 ParquetStatistics::ByteArray,
212 [u8]
213);
214make_stats_iterator!(
215 MaxByteArrayStatsIterator,
216 max_bytes_opt,
217 ParquetStatistics::ByteArray,
218 [u8]
219);
220make_stats_iterator!(
221 MinFixedLenByteArrayStatsIterator,
222 min_bytes_opt,
223 ParquetStatistics::FixedLenByteArray,
224 [u8]
225);
226make_stats_iterator!(
227 MaxFixedLenByteArrayStatsIterator,
228 max_bytes_opt,
229 ParquetStatistics::FixedLenByteArray,
230 [u8]
231);
232
233macro_rules! make_decimal_stats_iterator {
251 ($iterator_type:ident, $func:ident, $bytes_func:ident, $stat_value_type:ident, $convert_func: ident) => {
252 struct $iterator_type<'a, I>
253 where
254 I: Iterator<Item = Option<&'a ParquetStatistics>>,
255 {
256 iter: I,
257 }
258
259 impl<'a, I> $iterator_type<'a, I>
260 where
261 I: Iterator<Item = Option<&'a ParquetStatistics>>,
262 {
263 fn new(iter: I) -> Self {
264 Self { iter }
265 }
266 }
267
268 impl<'a, I> Iterator for $iterator_type<'a, I>
269 where
270 I: Iterator<Item = Option<&'a ParquetStatistics>>,
271 {
272 type Item = Option<$stat_value_type>;
273
274 fn next(&mut self) -> Option<Self::Item> {
275 let next = self.iter.next();
276 next.map(|x| {
277 x.and_then(|stats| match stats {
278 ParquetStatistics::Int32(s) => {
279 s.$func().map(|x| $stat_value_type::from(*x))
280 }
281 ParquetStatistics::Int64(s) => s
282 .$func()
283 .map(|x| $stat_value_type::try_from(*x).ok())
284 .flatten(),
285 ParquetStatistics::ByteArray(s) => s.$bytes_func().map($convert_func),
286 ParquetStatistics::FixedLenByteArray(s) => {
287 s.$bytes_func().map($convert_func)
288 }
289 _ => None,
290 })
291 })
292 }
293
294 fn size_hint(&self) -> (usize, Option<usize>) {
295 self.iter.size_hint()
296 }
297 }
298 };
299}
300
301make_decimal_stats_iterator!(
302 MinDecimal32StatsIterator,
303 min_opt,
304 min_bytes_opt,
305 i32,
306 from_bytes_to_i32
307);
308make_decimal_stats_iterator!(
309 MaxDecimal32StatsIterator,
310 max_opt,
311 max_bytes_opt,
312 i32,
313 from_bytes_to_i32
314);
315make_decimal_stats_iterator!(
316 MinDecimal64StatsIterator,
317 min_opt,
318 min_bytes_opt,
319 i64,
320 from_bytes_to_i64
321);
322make_decimal_stats_iterator!(
323 MaxDecimal64StatsIterator,
324 max_opt,
325 max_bytes_opt,
326 i64,
327 from_bytes_to_i64
328);
329make_decimal_stats_iterator!(
330 MinDecimal128StatsIterator,
331 min_opt,
332 min_bytes_opt,
333 i128,
334 from_bytes_to_i128
335);
336make_decimal_stats_iterator!(
337 MaxDecimal128StatsIterator,
338 max_opt,
339 max_bytes_opt,
340 i128,
341 from_bytes_to_i128
342);
343make_decimal_stats_iterator!(
344 MinDecimal256StatsIterator,
345 min_opt,
346 min_bytes_opt,
347 i256,
348 from_bytes_to_i256
349);
350make_decimal_stats_iterator!(
351 MaxDecimal256StatsIterator,
352 max_opt,
353 max_bytes_opt,
354 i256,
355 from_bytes_to_i256
356);
357
358macro_rules! get_statistics {
365 (Min, $data_type: ident, $iterator: ident, $physical_type: ident) => {
366 get_statistics!(
367 $data_type,
368 $iterator,
369 $physical_type,
370 MinBooleanStatsIterator,
371 MinInt32StatsIterator,
372 MinInt64StatsIterator,
373 MinFloatStatsIterator,
374 MinDoubleStatsIterator,
375 MinByteArrayStatsIterator,
376 MinFixedLenByteArrayStatsIterator,
377 MinDecimal32StatsIterator,
378 MinDecimal64StatsIterator,
379 MinDecimal128StatsIterator,
380 MinDecimal256StatsIterator,
381 min_statistics
382 )
383 };
384 (Max, $data_type: ident, $iterator: ident, $physical_type: ident) => {
385 get_statistics!(
386 $data_type,
387 $iterator,
388 $physical_type,
389 MaxBooleanStatsIterator,
390 MaxInt32StatsIterator,
391 MaxInt64StatsIterator,
392 MaxFloatStatsIterator,
393 MaxDoubleStatsIterator,
394 MaxByteArrayStatsIterator,
395 MaxFixedLenByteArrayStatsIterator,
396 MaxDecimal32StatsIterator,
397 MaxDecimal64StatsIterator,
398 MaxDecimal128StatsIterator,
399 MaxDecimal256StatsIterator,
400 max_statistics
401 )
402 };
403 (
404 $data_type: ident,
405 $iterator: ident,
406 $physical_type: ident,
407 $boolean_iter: ident,
408 $int32_iter: ident,
409 $int64_iter: ident,
410 $float_iter: ident,
411 $double_iter: ident,
412 $byte_array_iter: ident,
413 $fixed_len_byte_array_iter: ident,
414 $decimal32_iter: ident,
415 $decimal64_iter: ident,
416 $decimal128_iter: ident,
417 $decimal256_iter: ident,
418 $dictionary_statistics: ident
419 ) => {
420 match $data_type {
421 DataType::Boolean => Ok(Arc::new(BooleanArray::from_iter(
422 $boolean_iter::new($iterator).map(|x| x.copied()),
423 ))),
424 DataType::Int8 => Ok(Arc::new(Int8Array::from_iter(
425 $int32_iter::new($iterator).map(|x| {
426 x.and_then(|x| i8::try_from(*x).ok())
427 }),
428 ))),
429 DataType::Int16 => Ok(Arc::new(Int16Array::from_iter(
430 $int32_iter::new($iterator).map(|x| {
431 x.and_then(|x| i16::try_from(*x).ok())
432 }),
433 ))),
434 DataType::Int32 => Ok(Arc::new(Int32Array::from_iter(
435 $int32_iter::new($iterator).map(|x| x.copied()),
436 ))),
437 DataType::Int64 => Ok(Arc::new(Int64Array::from_iter(
438 $int64_iter::new($iterator).map(|x| x.copied()),
439 ))),
440 DataType::UInt8 => Ok(Arc::new(UInt8Array::from_iter(
441 $int32_iter::new($iterator).map(|x| {
442 x.and_then(|x| u8::try_from(*x).ok())
443 }),
444 ))),
445 DataType::UInt16 => Ok(Arc::new(UInt16Array::from_iter(
446 $int32_iter::new($iterator).map(|x| {
447 x.and_then(|x| u16::try_from(*x).ok())
448 }),
449 ))),
450 DataType::UInt32 => Ok(Arc::new(UInt32Array::from_iter(
451 $int32_iter::new($iterator).map(|x| x.map(|x| *x as u32)),
452 ))),
453 DataType::UInt64 => Ok(Arc::new(UInt64Array::from_iter(
454 $int64_iter::new($iterator).map(|x| x.map(|x| *x as u64)),
455 ))),
456 DataType::Float16 => Ok(Arc::new(Float16Array::from_iter(
457 $fixed_len_byte_array_iter::new($iterator).map(|x| x.and_then(|x| {
458 from_bytes_to_f16(x)
459 })),
460 ))),
461 DataType::Float32 => Ok(Arc::new(Float32Array::from_iter(
462 $float_iter::new($iterator).map(|x| x.copied()),
463 ))),
464 DataType::Float64 => Ok(Arc::new(Float64Array::from_iter(
465 $double_iter::new($iterator).map(|x| x.copied()),
466 ))),
467 DataType::Date32 => Ok(Arc::new(Date32Array::from_iter(
468 $int32_iter::new($iterator).map(|x| x.copied()),
469 ))),
470 DataType::Date64 if $physical_type == Some(PhysicalType::INT32) => Ok(Arc::new(Date64Array::from_iter(
471 $int32_iter::new($iterator)
472 .map(|x| x.map(|x| i64::from(*x) * 24 * 60 * 60 * 1000))))),
473 DataType::Date64 if $physical_type == Some(PhysicalType::INT64) => Ok(Arc::new(Date64Array::from_iter(
474 $int64_iter::new($iterator).map(|x| x.copied()),))),
475 DataType::Timestamp(unit, timezone) =>{
476 let iter = $int64_iter::new($iterator).map(|x| x.copied());
477 Ok(match unit {
478 TimeUnit::Second => Arc::new(TimestampSecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
479 TimeUnit::Millisecond => Arc::new(TimestampMillisecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
480 TimeUnit::Microsecond => Arc::new(TimestampMicrosecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
481 TimeUnit::Nanosecond => Arc::new(TimestampNanosecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
482 })
483 },
484 DataType::Time32(unit) => {
485 Ok(match unit {
486 TimeUnit::Second => Arc::new(Time32SecondArray::from_iter(
487 $int32_iter::new($iterator).map(|x| x.copied()),
488 )),
489 TimeUnit::Millisecond => Arc::new(Time32MillisecondArray::from_iter(
490 $int32_iter::new($iterator).map(|x| x.copied()),
491 )),
492 _ => {
493 let len = $iterator.count();
494 new_null_array($data_type, len)
496 }
497 })
498 },
499 DataType::Time64(unit) => {
500 Ok(match unit {
501 TimeUnit::Microsecond => Arc::new(Time64MicrosecondArray::from_iter(
502 $int64_iter::new($iterator).map(|x| x.copied()),
503 )),
504 TimeUnit::Nanosecond => Arc::new(Time64NanosecondArray::from_iter(
505 $int64_iter::new($iterator).map(|x| x.copied()),
506 )),
507 _ => {
508 let len = $iterator.count();
509 new_null_array($data_type, len)
511 }
512 })
513 },
514 DataType::Binary => Ok(Arc::new(BinaryArray::from_iter(
515 $byte_array_iter::new($iterator)
516 ))),
517 DataType::LargeBinary => Ok(Arc::new(LargeBinaryArray::from_iter(
518 $byte_array_iter::new($iterator)
519 ))),
520 DataType::Utf8 => {
521 let iterator = $byte_array_iter::new($iterator);
522 let mut builder = StringBuilder::new();
523 for x in iterator {
524 let Some(x) = x else {
525 builder.append_null(); continue;
527 };
528
529 let Ok(x) = std::str::from_utf8(x) else {
530 builder.append_null();
531 continue;
532 };
533
534 builder.append_value(x);
535 }
536 Ok(Arc::new(builder.finish()))
537 },
538 DataType::LargeUtf8 => {
539 let iterator = $byte_array_iter::new($iterator);
540 let mut builder = LargeStringBuilder::new();
541 for x in iterator {
542 let Some(x) = x else {
543 builder.append_null(); continue;
545 };
546
547 let Ok(x) = std::str::from_utf8(x) else {
548 builder.append_null();
549 continue;
550 };
551
552 builder.append_value(x);
553 }
554 Ok(Arc::new(builder.finish()))
555 },
556 DataType::FixedSizeBinary(size) => {
557 let iterator = $fixed_len_byte_array_iter::new($iterator);
558 let mut builder = FixedSizeBinaryBuilder::new(*size);
559 for x in iterator {
560 let Some(x) = x else {
561 builder.append_null(); continue;
563 };
564
565 if x.len().try_into() != Ok(*size){
567 builder.append_null();
568 continue;
569 }
570
571 builder.append_value(x).expect("ensure to append successfully here, because size have been checked before");
572 }
573 Ok(Arc::new(builder.finish()))
574 },
575 DataType::Decimal32(precision, scale) => {
576 let arr = Decimal32Array::from_iter(
577 $decimal32_iter::new($iterator)
578 ).with_precision_and_scale(*precision, *scale)?;
579 Ok(Arc::new(arr))
580 },
581 DataType::Decimal64(precision, scale) => {
582 let arr = Decimal64Array::from_iter(
583 $decimal64_iter::new($iterator)
584 ).with_precision_and_scale(*precision, *scale)?;
585 Ok(Arc::new(arr))
586 },
587 DataType::Decimal128(precision, scale) => {
588 let arr = Decimal128Array::from_iter(
589 $decimal128_iter::new($iterator)
590 ).with_precision_and_scale(*precision, *scale)?;
591 Ok(Arc::new(arr))
592 },
593 DataType::Decimal256(precision, scale) => {
594 let arr = Decimal256Array::from_iter(
595 $decimal256_iter::new($iterator)
596 ).with_precision_and_scale(*precision, *scale)?;
597 Ok(Arc::new(arr))
598 },
599 DataType::Dictionary(_, value_type) => {
600 $dictionary_statistics(value_type, $iterator, $physical_type)
601 },
602 DataType::Utf8View => {
603 let iterator = $byte_array_iter::new($iterator);
604 let mut builder = StringViewBuilder::new();
605 for x in iterator {
606 let Some(x) = x else {
607 builder.append_null(); continue;
609 };
610
611 let Ok(x) = std::str::from_utf8(x) else {
612 builder.append_null();
613 continue;
614 };
615
616 builder.append_value(x);
617 }
618 Ok(Arc::new(builder.finish()))
619 },
620 DataType::BinaryView => {
621 let iterator = $byte_array_iter::new($iterator);
622 let mut builder = BinaryViewBuilder::new();
623 for x in iterator {
624 let Some(x) = x else {
625 builder.append_null(); continue;
627 };
628
629 builder.append_value(x);
630 }
631 Ok(Arc::new(builder.finish()))
632 }
633
634 DataType::Map(_,_) |
635 DataType::Duration(_) |
636 DataType::Interval(_) |
637 DataType::Date64 | DataType::Null |
639 DataType::List(_) |
640 DataType::ListView(_) |
641 DataType::FixedSizeList(_, _) |
642 DataType::LargeList(_) |
643 DataType::LargeListView(_) |
644 DataType::Struct(_) |
645 DataType::Union(_, _) |
646 DataType::RunEndEncoded(_, _) => {
647 let len = $iterator.count();
648 Ok(new_null_array($data_type, len))
650 }
651 }
652 };
653}
654
655macro_rules! get_data_page_statistics {
656 (Min, $data_type: ident, $iterator: ident, $physical_type: ident) => {
657 get_data_page_statistics!(
658 $data_type,
659 $iterator,
660 $physical_type,
661 min_values_iter,
662 min_page_statistics
663 )
664 };
665 (Max, $data_type: ident, $iterator: ident, $physical_type: ident) => {
666 get_data_page_statistics!(
667 $data_type,
668 $iterator,
669 $physical_type,
670 max_values_iter,
671 max_page_statistics
672 )
673 };
674 (
675 $data_type: ident,
676 $iterator: ident,
677 $physical_type: ident,
678 $values_iter: ident,
679 $page_statistics: ident
680 ) => {{
681 let chunks: Vec<(usize, &ColumnIndexMetaData)> = $iterator.collect();
682 let capacity: usize = chunks.iter().map(|c| c.0).sum();
683 match $data_type {
684 DataType::Boolean => {
685 let mut b = BooleanBuilder::with_capacity(capacity);
686 for (len, index) in chunks {
687 match index {
688 ColumnIndexMetaData::BOOLEAN(index) => {
689 for val in index.$values_iter() {
690 b.append_option(val.copied());
691 }
692 }
693 _ => b.append_nulls(len),
694 }
695 }
696 Ok(Arc::new(b.finish()))
697 },
698 DataType::UInt8 => {
699 let mut b = UInt8Builder::with_capacity(capacity);
700 for (len, index) in chunks {
701 match index {
702 ColumnIndexMetaData::INT32(index) => {
703 b.extend_from_iter_option(
704 index.$values_iter()
705 .map(|val| val.and_then(|&x| u8::try_from(x).ok())),
706 );
707 }
708 _ => b.append_nulls(len),
709 }
710 }
711 Ok(Arc::new(b.finish()))
712 },
713 DataType::UInt16 => {
714 let mut b = UInt16Builder::with_capacity(capacity);
715 for (len, index) in chunks {
716 match index {
717 ColumnIndexMetaData::INT32(index) => {
718 b.extend_from_iter_option(
719 index.$values_iter()
720 .map(|val| val.and_then(|&x| u16::try_from(x).ok())),
721 );
722 }
723 _ => b.append_nulls(len),
724 }
725 }
726 Ok(Arc::new(b.finish()))
727 },
728 DataType::UInt32 => {
729 let mut b = UInt32Builder::with_capacity(capacity);
730 for (len, index) in chunks {
731 match index {
732 ColumnIndexMetaData::INT32(index) => {
733 b.extend_from_iter_option(
734 index.$values_iter()
735 .map(|val| val.map(|&x| x as u32)),
736 );
737 }
738 _ => b.append_nulls(len),
739 }
740 }
741 Ok(Arc::new(b.finish()))
742 },
743 DataType::UInt64 => {
744 let mut b = UInt64Builder::with_capacity(capacity);
745 for (len, index) in chunks {
746 match index {
747 ColumnIndexMetaData::INT64(index) => {
748 b.extend_from_iter_option(
749 index.$values_iter()
750 .map(|val| val.map(|&x| x as u64)),
751 );
752 }
753 _ => b.append_nulls(len),
754 }
755 }
756 Ok(Arc::new(b.finish()))
757 },
758 DataType::Int8 => {
759 let mut b = Int8Builder::with_capacity(capacity);
760 for (len, index) in chunks {
761 match index {
762 ColumnIndexMetaData::INT32(index) => {
763 b.extend_from_iter_option(
764 index.$values_iter()
765 .map(|val| val.and_then(|&x| i8::try_from(x).ok())),
766 );
767 }
768 _ => b.append_nulls(len),
769 }
770 }
771 Ok(Arc::new(b.finish()))
772 },
773 DataType::Int16 => {
774 let mut b = Int16Builder::with_capacity(capacity);
775 for (len, index) in chunks {
776 match index {
777 ColumnIndexMetaData::INT32(index) => {
778 b.extend_from_iter_option(
779 index.$values_iter()
780 .map(|val| val.and_then(|&x| i16::try_from(x).ok())),
781 );
782 }
783 _ => b.append_nulls(len),
784 }
785 }
786 Ok(Arc::new(b.finish()))
787 },
788 DataType::Int32 => {
789 let mut b = Int32Builder::with_capacity(capacity);
790 for (len, index) in chunks {
791 match index {
792 ColumnIndexMetaData::INT32(index) => {
793 b.extend_from_iter_option(
794 index.$values_iter()
795 .map(|val| val.copied()),
796 );
797 }
798 _ => b.append_nulls(len),
799 }
800 }
801 Ok(Arc::new(b.finish()))
802 },
803 DataType::Int64 => {
804 let mut b = Int64Builder::with_capacity(capacity);
805 for (len, index) in chunks {
806 match index {
807 ColumnIndexMetaData::INT64(index) => {
808 b.extend_from_iter_option(
809 index.$values_iter()
810 .map(|val| val.copied()),
811 );
812 }
813 _ => b.append_nulls(len),
814 }
815 }
816 Ok(Arc::new(b.finish()))
817 },
818 DataType::Float16 => {
819 let mut b = Float16Builder::with_capacity(capacity);
820 for (len, index) in chunks {
821 match index {
822 ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index) => {
823 b.extend_from_iter_option(
824 index.$values_iter()
825 .map(|val| val.and_then(|x| from_bytes_to_f16(x))),
826 );
827 }
828 _ => b.append_nulls(len),
829 }
830 }
831 Ok(Arc::new(b.finish()))
832 },
833 DataType::Float32 => {
834 let mut b = Float32Builder::with_capacity(capacity);
835 for (len, index) in chunks {
836 match index {
837 ColumnIndexMetaData::FLOAT(index) => {
838 b.extend_from_iter_option(
839 index.$values_iter()
840 .map(|val| val.copied()),
841 );
842 }
843 _ => b.append_nulls(len),
844 }
845 }
846 Ok(Arc::new(b.finish()))
847 },
848 DataType::Float64 => {
849 let mut b = Float64Builder::with_capacity(capacity);
850 for (len, index) in chunks {
851 match index {
852 ColumnIndexMetaData::DOUBLE(index) => {
853 b.extend_from_iter_option(
854 index.$values_iter()
855 .map(|val| val.copied()),
856 );
857 }
858 _ => b.append_nulls(len),
859 }
860 }
861 Ok(Arc::new(b.finish()))
862 },
863 DataType::Binary => {
864 let mut b = BinaryBuilder::with_capacity(capacity, capacity * 10);
865 for (len, index) in chunks {
866 match index {
867 ColumnIndexMetaData::BYTE_ARRAY(index) => {
868 for val in index.$values_iter() {
869 b.append_option(val.map(|x| x.as_ref()));
870 }
871 }
872 _ => b.append_nulls(len),
873 }
874 }
875 Ok(Arc::new(b.finish()))
876 },
877 DataType::LargeBinary => {
878 let mut b = LargeBinaryBuilder::with_capacity(capacity, capacity * 10);
879 for (len, index) in chunks {
880 match index {
881 ColumnIndexMetaData::BYTE_ARRAY(index) => {
882 for val in index.$values_iter() {
883 b.append_option(val.map(|x| x.as_ref()));
884 }
885 }
886 _ => b.append_nulls(len),
887 }
888 }
889 Ok(Arc::new(b.finish()))
890 },
891 DataType::Utf8 => {
892 let mut b = StringBuilder::with_capacity(capacity, capacity * 10);
893 for (len, index) in chunks {
894 match index {
895 ColumnIndexMetaData::BYTE_ARRAY(index) => {
896 for val in index.$values_iter() {
897 match val {
898 Some(x) => match std::str::from_utf8(x.as_ref()) {
899 Ok(s) => b.append_value(s),
900 _ => b.append_null(),
901 }
902 None => b.append_null(),
903 }
904 }
905 }
906 _ => b.append_nulls(len),
907 }
908 }
909 Ok(Arc::new(b.finish()))
910 },
911 DataType::LargeUtf8 => {
912 let mut b = LargeStringBuilder::with_capacity(capacity, capacity * 10);
913 for (len, index) in chunks {
914 match index {
915 ColumnIndexMetaData::BYTE_ARRAY(index) => {
916 for val in index.$values_iter() {
917 match val {
918 Some(x) => match std::str::from_utf8(x.as_ref()) {
919 Ok(s) => b.append_value(s),
920 _ => b.append_null(),
921 }
922 None => b.append_null(),
923 }
924 }
925 }
926 _ => b.append_nulls(len),
927 }
928 }
929 Ok(Arc::new(b.finish()))
930 },
931 DataType::Dictionary(_, value_type) => {
932 $page_statistics(value_type, chunks.into_iter(), $physical_type)
933 },
934 DataType::Timestamp(unit, timezone) => {
935 match unit {
936 TimeUnit::Second => {
937 let mut b = TimestampSecondBuilder::with_capacity(capacity);
938 for (len, index) in chunks {
939 match index {
940 ColumnIndexMetaData::INT64(index) => {
941 b.extend_from_iter_option(
942 index.$values_iter()
943 .map(|val| val.copied()),
944 );
945 }
946 _ => b.append_nulls(len),
947 }
948 }
949 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
950 }
951 TimeUnit::Millisecond => {
952 let mut b = TimestampMillisecondBuilder::with_capacity(capacity);
953 for (len, index) in chunks {
954 match index {
955 ColumnIndexMetaData::INT64(index) => {
956 b.extend_from_iter_option(
957 index.$values_iter()
958 .map(|val| val.copied()),
959 );
960 }
961 _ => b.append_nulls(len),
962 }
963 }
964 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
965 }
966 TimeUnit::Microsecond => {
967 let mut b = TimestampMicrosecondBuilder::with_capacity(capacity);
968 for (len, index) in chunks {
969 match index {
970 ColumnIndexMetaData::INT64(index) => {
971 b.extend_from_iter_option(
972 index.$values_iter()
973 .map(|val| val.copied()),
974 );
975 }
976 _ => b.append_nulls(len),
977 }
978 }
979 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
980 }
981 TimeUnit::Nanosecond => {
982 let mut b = TimestampNanosecondBuilder::with_capacity(capacity);
983 for (len, index) in chunks {
984 match index {
985 ColumnIndexMetaData::INT64(index) => {
986 b.extend_from_iter_option(
987 index.$values_iter()
988 .map(|val| val.copied()),
989 );
990 }
991 _ => b.append_nulls(len),
992 }
993 }
994 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
995 }
996 }
997 },
998 DataType::Date32 => {
999 let mut b = Date32Builder::with_capacity(capacity);
1000 for (len, index) in chunks {
1001 match index {
1002 ColumnIndexMetaData::INT32(index) => {
1003 b.extend_from_iter_option(
1004 index.$values_iter()
1005 .map(|val| val.copied()),
1006 );
1007 }
1008 _ => b.append_nulls(len),
1009 }
1010 }
1011 Ok(Arc::new(b.finish()))
1012 },
1013 DataType::Date64 if $physical_type == Some(PhysicalType::INT32)=> {
1014 let mut b = Date64Builder::with_capacity(capacity);
1015 for (len, index) in chunks {
1016 match index {
1017 ColumnIndexMetaData::INT32(index) => {
1018 b.extend_from_iter_option(
1019 index.$values_iter()
1020 .map(|val| val.map(|&x| (x as i64) * 24 * 60 * 60 * 1000)),
1021 );
1022 }
1023 _ => b.append_nulls(len),
1024 }
1025 }
1026 Ok(Arc::new(b.finish()))
1027 },
1028 DataType::Date64 if $physical_type == Some(PhysicalType::INT64) => {
1029 let mut b = Date64Builder::with_capacity(capacity);
1030 for (len, index) in chunks {
1031 match index {
1032 ColumnIndexMetaData::INT64(index) => {
1033 b.extend_from_iter_option(
1034 index.$values_iter()
1035 .map(|val| val.copied()),
1036 );
1037 }
1038 _ => b.append_nulls(len),
1039 }
1040 }
1041 Ok(Arc::new(b.finish()))
1042 },
1043 DataType::Decimal32(precision, scale) => {
1044 let mut b = Decimal32Builder::with_capacity(capacity);
1045 for (len, index) in chunks {
1046 match index {
1047 ColumnIndexMetaData::INT32(index) => {
1048 b.extend_from_iter_option(
1049 index.$values_iter()
1050 .map(|val| val.copied()),
1051 );
1052 }
1053 ColumnIndexMetaData::INT64(index) => {
1054 b.extend_from_iter_option(
1055 index.$values_iter()
1056 .map(|val| val.and_then(|&x| i32::try_from(x).ok())),
1057 );
1058 }
1059 ColumnIndexMetaData::BYTE_ARRAY(index) => {
1060 b.extend_from_iter_option(
1061 index.$values_iter()
1062 .map(|val| val.map(|x| from_bytes_to_i32(x.as_ref()))),
1063 );
1064 }
1065 ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index) => {
1066 b.extend_from_iter_option(
1067 index.$values_iter()
1068 .map(|val| val.map(|x| from_bytes_to_i32(x.as_ref()))),
1069 );
1070 }
1071 _ => b.append_nulls(len),
1072 }
1073 }
1074 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1075 },
1076 DataType::Decimal64(precision, scale) => {
1077 let mut b = Decimal64Builder::with_capacity(capacity);
1078 for (len, index) in chunks {
1079 match index {
1080 ColumnIndexMetaData::INT32(index) => {
1081 b.extend_from_iter_option(
1082 index.$values_iter()
1083 .map(|val| val.map(|x| *x as i64)),
1084 );
1085 }
1086 ColumnIndexMetaData::INT64(index) => {
1087 b.extend_from_iter_option(
1088 index.$values_iter()
1089 .map(|val| val.copied()),
1090 );
1091 }
1092 ColumnIndexMetaData::BYTE_ARRAY(index) => {
1093 b.extend_from_iter_option(
1094 index.$values_iter()
1095 .map(|val| val.map(|x| from_bytes_to_i64(x.as_ref()))),
1096 );
1097 }
1098 ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index) => {
1099 b.extend_from_iter_option(
1100 index.$values_iter()
1101 .map(|val| val.map(|x| from_bytes_to_i64(x.as_ref()))),
1102 );
1103 }
1104 _ => b.append_nulls(len),
1105 }
1106 }
1107 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1108 },
1109 DataType::Decimal128(precision, scale) => {
1110 let mut b = Decimal128Array::builder(capacity);
1111 for (len, index) in chunks {
1112 match index {
1113 ColumnIndexMetaData::INT32(index) => {
1114 b.extend_from_iter_option(
1115 index.$values_iter()
1116 .map(|val| val.map(|x| *x as i128)),
1117 );
1118 }
1119 ColumnIndexMetaData::INT64(index) => {
1120 b.extend_from_iter_option(
1121 index.$values_iter()
1122 .map(|val| val.map(|x| *x as i128)),
1123 );
1124 }
1125 ColumnIndexMetaData::BYTE_ARRAY(index) => {
1126 b.extend_from_iter_option(
1127 index.$values_iter()
1128 .map(|val| val.map(|x| from_bytes_to_i128(x.as_ref()))),
1129 );
1130 }
1131 ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index) => {
1132 b.extend_from_iter_option(
1133 index.$values_iter()
1134 .map(|val| val.map(|x| from_bytes_to_i128(x.as_ref()))),
1135 );
1136 }
1137 _ => b.append_nulls(len),
1138 }
1139 }
1140 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1141 },
1142 DataType::Decimal256(precision, scale) => {
1143 let mut b = Decimal256Array::builder(capacity);
1144 for (len, index) in chunks {
1145 match index {
1146 ColumnIndexMetaData::INT32(index) => {
1147 b.extend_from_iter_option(
1148 index.$values_iter()
1149 .map(|val| val.map(|x| i256::from_i128(*x as i128))),
1150 );
1151 }
1152 ColumnIndexMetaData::INT64(index) => {
1153 b.extend_from_iter_option(
1154 index.$values_iter()
1155 .map(|val| val.map(|x| i256::from_i128(*x as i128))),
1156 );
1157 }
1158 ColumnIndexMetaData::BYTE_ARRAY(index) => {
1159 b.extend_from_iter_option(
1160 index.$values_iter()
1161 .map(|val| val.map(|x| from_bytes_to_i256(x.as_ref()))),
1162 );
1163 }
1164 ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index) => {
1165 b.extend_from_iter_option(
1166 index.$values_iter()
1167 .map(|val| val.map(|x| from_bytes_to_i256(x.as_ref()))),
1168 );
1169 }
1170 _ => b.append_nulls(len),
1171 }
1172 }
1173 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1174 },
1175 DataType::Time32(unit) => {
1176 match unit {
1177 TimeUnit::Second => {
1178 let mut b = Time32SecondBuilder::with_capacity(capacity);
1179 for (len, index) in chunks {
1180 match index {
1181 ColumnIndexMetaData::INT32(index) => {
1182 b.extend_from_iter_option(
1183 index.$values_iter()
1184 .map(|val| val.copied()),
1185 );
1186 }
1187 _ => b.append_nulls(len),
1188 }
1189 }
1190 Ok(Arc::new(b.finish()))
1191 }
1192 TimeUnit::Millisecond => {
1193 let mut b = Time32MillisecondBuilder::with_capacity(capacity);
1194 for (len, index) in chunks {
1195 match index {
1196 ColumnIndexMetaData::INT32(index) => {
1197 b.extend_from_iter_option(
1198 index.$values_iter()
1199 .map(|val| val.copied()),
1200 );
1201 }
1202 _ => b.append_nulls(len),
1203 }
1204 }
1205 Ok(Arc::new(b.finish()))
1206 }
1207 _ => {
1208 Ok(new_null_array($data_type, capacity))
1209 }
1210 }
1211 }
1212 DataType::Time64(unit) => {
1213 match unit {
1214 TimeUnit::Microsecond => {
1215 let mut b = Time64MicrosecondBuilder::with_capacity(capacity);
1216 for (len, index) in chunks {
1217 match index {
1218 ColumnIndexMetaData::INT64(index) => {
1219 b.extend_from_iter_option(
1220 index.$values_iter()
1221 .map(|val| val.copied()),
1222 );
1223 }
1224 _ => b.append_nulls(len),
1225 }
1226 }
1227 Ok(Arc::new(b.finish()))
1228 }
1229 TimeUnit::Nanosecond => {
1230 let mut b = Time64NanosecondBuilder::with_capacity(capacity);
1231 for (len, index) in chunks {
1232 match index {
1233 ColumnIndexMetaData::INT64(index) => {
1234 b.extend_from_iter_option(
1235 index.$values_iter()
1236 .map(|val| val.copied()),
1237 );
1238 }
1239 _ => b.append_nulls(len),
1240 }
1241 }
1242 Ok(Arc::new(b.finish()))
1243 }
1244 _ => {
1245 Ok(new_null_array($data_type, capacity))
1246 }
1247 }
1248 },
1249 DataType::FixedSizeBinary(size) => {
1250 let mut b = FixedSizeBinaryBuilder::with_capacity(capacity, *size);
1251 for (len, index) in chunks {
1252 match index {
1253 ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index) => {
1254 for val in index.$values_iter() {
1255 match val {
1256 Some(v) => {
1257 if v.len() == *size as usize {
1258 let _ = b.append_value(v.as_ref())?;
1259 } else {
1260 b.append_null();
1261 }
1262 }
1263 None => b.append_null(),
1264 }
1265 }
1266 }
1267 _ => b.append_nulls(len),
1268 }
1269 }
1270 Ok(Arc::new(b.finish()))
1271 },
1272 DataType::Utf8View => {
1273 let mut b = StringViewBuilder::with_capacity(capacity);
1274 for (len, index) in chunks {
1275 match index {
1276 ColumnIndexMetaData::BYTE_ARRAY(index) => {
1277 for val in index.$values_iter() {
1278 match val {
1279 Some(x) => match std::str::from_utf8(x.as_ref()) {
1280 Ok(s) => b.append_value(s),
1281 _ => b.append_null(),
1282 }
1283 None => b.append_null(),
1284 }
1285 }
1286 }
1287 _ => {
1288 for _ in 0..len { b.append_null(); }
1289 }
1290 }
1291 }
1292 Ok(Arc::new(b.finish()))
1293 },
1294 DataType::BinaryView => {
1295 let mut b = BinaryViewBuilder::with_capacity(capacity);
1296 for (len, index) in chunks {
1297 match index {
1298 ColumnIndexMetaData::BYTE_ARRAY(index) => {
1299 for val in index.$values_iter() {
1300 match val {
1301 Some(v) => b.append_value(v.as_ref()),
1302 None => b.append_null(),
1303 }
1304 }
1305 }
1306 _ => {
1307 for _ in 0..len { b.append_null(); }
1308 }
1309 }
1310 }
1311 Ok(Arc::new(b.finish()))
1312 },
1313 DataType::Date64 | DataType::Null |
1315 DataType::Duration(_) |
1316 DataType::Interval(_) |
1317 DataType::List(_) |
1318 DataType::ListView(_) |
1319 DataType::FixedSizeList(_, _) |
1320 DataType::LargeList(_) |
1321 DataType::LargeListView(_) |
1322 DataType::Struct(_) |
1323 DataType::Union(_, _) |
1324 DataType::Map(_, _) |
1325 DataType::RunEndEncoded(_, _) => {
1326 Ok(new_null_array($data_type, capacity))
1328 },
1329 }
1330 }
1331 };
1332}
1333fn min_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
1338 data_type: &DataType,
1339 iterator: I,
1340 physical_type: Option<PhysicalType>,
1341) -> Result<ArrayRef> {
1342 get_statistics!(Min, data_type, iterator, physical_type)
1343}
1344
1345fn max_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
1349 data_type: &DataType,
1350 iterator: I,
1351 physical_type: Option<PhysicalType>,
1352) -> Result<ArrayRef> {
1353 get_statistics!(Max, data_type, iterator, physical_type)
1354}
1355
1356pub(crate) fn min_page_statistics<'a, I>(
1359 data_type: &DataType,
1360 iterator: I,
1361 physical_type: Option<PhysicalType>,
1362) -> Result<ArrayRef>
1363where
1364 I: Iterator<Item = (usize, &'a ColumnIndexMetaData)>,
1365{
1366 get_data_page_statistics!(Min, data_type, iterator, physical_type)
1367}
1368
1369pub(crate) fn max_page_statistics<'a, I>(
1372 data_type: &DataType,
1373 iterator: I,
1374 physical_type: Option<PhysicalType>,
1375) -> Result<ArrayRef>
1376where
1377 I: Iterator<Item = (usize, &'a ColumnIndexMetaData)>,
1378{
1379 get_data_page_statistics!(Max, data_type, iterator, physical_type)
1380}
1381
1382pub(crate) fn null_counts_page_statistics<'a, I>(iterator: I) -> Result<UInt64Array>
1387where
1388 I: Iterator<Item = (usize, &'a ColumnIndexMetaData)>,
1389{
1390 let chunks: Vec<_> = iterator.collect();
1391 let total_capacity: usize = chunks.iter().map(|(len, _)| *len).sum();
1392 let mut values = Vec::with_capacity(total_capacity);
1393 let mut nulls = NullBufferBuilder::new(total_capacity);
1394 for (len, index) in chunks {
1395 match index.null_counts() {
1396 Some(counts) => {
1397 values.extend(counts.iter().map(|&x| x as u64));
1398 nulls.append_n_non_nulls(len);
1399 }
1400 None => {
1401 values.resize(values.len() + len, 0);
1402 nulls.append_n_nulls(len);
1403 }
1404 }
1405 }
1406 let null_buffer = nulls.build();
1407 let array = UInt64Array::new(values.into(), null_buffer);
1408 Ok(array)
1409}
1410
1411#[derive(Debug)]
1434pub struct StatisticsConverter<'a> {
1435 parquet_column_index: Option<usize>,
1437 arrow_field: &'a Field,
1439 missing_null_counts_as_zero: bool,
1441 physical_type: Option<PhysicalType>,
1443}
1444
1445impl<'a> StatisticsConverter<'a> {
1446 pub fn parquet_column_index(&self) -> Option<usize> {
1451 self.parquet_column_index
1452 }
1453
1454 pub fn arrow_field(&self) -> &'a Field {
1456 self.arrow_field
1457 }
1458
1459 pub fn with_missing_null_counts_as_zero(mut self, missing_null_counts_as_zero: bool) -> Self {
1472 self.missing_null_counts_as_zero = missing_null_counts_as_zero;
1473 self
1474 }
1475
1476 pub fn row_group_row_counts<I>(&self, metadatas: I) -> Result<Option<UInt64Array>>
1507 where
1508 I: IntoIterator<Item = &'a RowGroupMetaData>,
1509 {
1510 let Some(_) = self.parquet_column_index else {
1511 return Ok(None);
1512 };
1513
1514 let mut builder = UInt64Array::builder(10);
1515 for metadata in metadatas.into_iter() {
1516 let row_count = metadata.num_rows();
1517 let row_count: u64 = row_count.try_into().map_err(|e| {
1518 arrow_err!(format!(
1519 "Parquet row count {row_count} too large to convert to u64: {e}"
1520 ))
1521 })?;
1522 builder.append_value(row_count);
1523 }
1524 Ok(Some(builder.finish()))
1525 }
1526
1527 pub fn try_new<'b>(
1542 column_name: &'b str,
1543 arrow_schema: &'a Schema,
1544 parquet_schema: &'a SchemaDescriptor,
1545 ) -> Result<Self> {
1546 let Some((_idx, arrow_field)) = arrow_schema.column_with_name(column_name) else {
1548 return Err(arrow_err!(format!(
1549 "Column '{}' not found in schema for statistics conversion",
1550 column_name
1551 )));
1552 };
1553
1554 let parquet_index = match parquet_column(parquet_schema, arrow_schema, column_name) {
1556 Some((parquet_idx, matched_field)) => {
1557 if matched_field.as_ref() != arrow_field {
1559 return Err(arrow_err!(format!(
1560 "Matched column '{:?}' does not match original matched column '{:?}'",
1561 matched_field, arrow_field
1562 )));
1563 }
1564 Some(parquet_idx)
1565 }
1566 None => None,
1567 };
1568
1569 Ok(Self {
1570 parquet_column_index: parquet_index,
1571 arrow_field,
1572 missing_null_counts_as_zero: true,
1573 physical_type: parquet_index.map(|idx| parquet_schema.column(idx).physical_type()),
1574 })
1575 }
1576
1577 pub fn from_column_index(
1599 parquet_column_index: usize,
1600 arrow_field: &'a Field,
1601 parquet_schema: &'a SchemaDescriptor,
1602 ) -> Result<Self> {
1603 if parquet_column_index >= parquet_schema.columns().len() {
1604 return Err(arrow_err!(format!(
1605 "Parquet column index {} out of bounds, column count {}",
1606 parquet_column_index,
1607 parquet_schema.columns().len()
1608 )));
1609 }
1610
1611 let physical_type = parquet_schema.column(parquet_column_index).physical_type();
1612
1613 Ok(Self {
1614 parquet_column_index: Some(parquet_column_index),
1615 arrow_field,
1616 missing_null_counts_as_zero: true,
1617 physical_type: Some(physical_type),
1618 })
1619 }
1620
1621 pub fn row_group_mins<I>(&self, metadatas: I) -> Result<ArrayRef>
1665 where
1666 I: IntoIterator<Item = &'a RowGroupMetaData>,
1667 {
1668 let data_type = self.arrow_field.data_type();
1669
1670 let Some(parquet_index) = self.parquet_column_index else {
1671 return Ok(self.make_null_array(data_type, metadatas));
1672 };
1673
1674 let iter = metadatas
1675 .into_iter()
1676 .map(|x| x.column(parquet_index).statistics());
1677 min_statistics(data_type, iter, self.physical_type)
1678 }
1679
1680 pub fn row_group_maxes<I>(&self, metadatas: I) -> Result<ArrayRef>
1684 where
1685 I: IntoIterator<Item = &'a RowGroupMetaData>,
1686 {
1687 let data_type = self.arrow_field.data_type();
1688
1689 let Some(parquet_index) = self.parquet_column_index else {
1690 return Ok(self.make_null_array(data_type, metadatas));
1691 };
1692
1693 let iter = metadatas
1694 .into_iter()
1695 .map(|x| x.column(parquet_index).statistics());
1696 max_statistics(data_type, iter, self.physical_type)
1697 }
1698
1699 pub fn row_group_is_max_value_exact<I>(&self, metadatas: I) -> Result<BooleanArray>
1703 where
1704 I: IntoIterator<Item = &'a RowGroupMetaData>,
1705 {
1706 let Some(parquet_index) = self.parquet_column_index else {
1707 let num_row_groups = metadatas.into_iter().count();
1708 return Ok(BooleanArray::from_iter(std::iter::repeat_n(
1709 None,
1710 num_row_groups,
1711 )));
1712 };
1713
1714 let is_max_value_exact = metadatas
1715 .into_iter()
1716 .map(|x| x.column(parquet_index).statistics())
1717 .map(|s| s.map(|s| s.max_is_exact()));
1718 Ok(BooleanArray::from_iter(is_max_value_exact))
1719 }
1720
1721 pub fn row_group_is_min_value_exact<I>(&self, metadatas: I) -> Result<BooleanArray>
1725 where
1726 I: IntoIterator<Item = &'a RowGroupMetaData>,
1727 {
1728 let Some(parquet_index) = self.parquet_column_index else {
1729 let num_row_groups = metadatas.into_iter().count();
1730 return Ok(BooleanArray::from_iter(std::iter::repeat_n(
1731 None,
1732 num_row_groups,
1733 )));
1734 };
1735
1736 let is_min_value_exact = metadatas
1737 .into_iter()
1738 .map(|x| x.column(parquet_index).statistics())
1739 .map(|s| s.map(|s| s.min_is_exact()));
1740 Ok(BooleanArray::from_iter(is_min_value_exact))
1741 }
1742
1743 pub fn row_group_null_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
1747 where
1748 I: IntoIterator<Item = &'a RowGroupMetaData>,
1749 {
1750 let Some(parquet_index) = self.parquet_column_index else {
1751 let num_row_groups = metadatas.into_iter().count();
1752 return Ok(UInt64Array::from_iter(std::iter::repeat_n(
1753 None,
1754 num_row_groups,
1755 )));
1756 };
1757
1758 let null_counts = metadatas
1759 .into_iter()
1760 .map(|x| x.column(parquet_index).statistics())
1761 .map(|s| {
1762 s.and_then(|s| {
1763 if self.missing_null_counts_as_zero {
1764 Some(s.null_count_opt().unwrap_or(0))
1765 } else {
1766 s.null_count_opt()
1767 }
1768 })
1769 });
1770 Ok(UInt64Array::from_iter(null_counts))
1771 }
1772
1773 pub fn data_page_mins<I>(
1825 &self,
1826 column_page_index: &ParquetColumnIndex,
1827 column_offset_index: &ParquetOffsetIndex,
1828 row_group_indices: I,
1829 ) -> Result<ArrayRef>
1830 where
1831 I: IntoIterator<Item = &'a usize>,
1832 {
1833 let data_type = self.arrow_field.data_type();
1834
1835 let Some(parquet_index) = self.parquet_column_index else {
1836 return Ok(self.make_null_array(data_type, row_group_indices));
1837 };
1838
1839 let iter = row_group_indices.into_iter().map(|rg_index| {
1840 let column_page_index_per_row_group_per_column =
1841 &column_page_index[*rg_index][parquet_index];
1842 let num_data_pages = &column_offset_index[*rg_index][parquet_index]
1843 .page_locations()
1844 .len();
1845
1846 (*num_data_pages, column_page_index_per_row_group_per_column)
1847 });
1848
1849 min_page_statistics(data_type, iter, self.physical_type)
1850 }
1851
1852 pub fn data_page_maxes<I>(
1856 &self,
1857 column_page_index: &ParquetColumnIndex,
1858 column_offset_index: &ParquetOffsetIndex,
1859 row_group_indices: I,
1860 ) -> Result<ArrayRef>
1861 where
1862 I: IntoIterator<Item = &'a usize>,
1863 {
1864 let data_type = self.arrow_field.data_type();
1865
1866 let Some(parquet_index) = self.parquet_column_index else {
1867 return Ok(self.make_null_array(data_type, row_group_indices));
1868 };
1869
1870 let iter = row_group_indices.into_iter().map(|rg_index| {
1871 let column_page_index_per_row_group_per_column =
1872 &column_page_index[*rg_index][parquet_index];
1873 let num_data_pages = &column_offset_index[*rg_index][parquet_index]
1874 .page_locations()
1875 .len();
1876
1877 (*num_data_pages, column_page_index_per_row_group_per_column)
1878 });
1879
1880 max_page_statistics(data_type, iter, self.physical_type)
1881 }
1882
1883 pub fn data_page_null_counts<I>(
1887 &self,
1888 column_page_index: &ParquetColumnIndex,
1889 column_offset_index: &ParquetOffsetIndex,
1890 row_group_indices: I,
1891 ) -> Result<UInt64Array>
1892 where
1893 I: IntoIterator<Item = &'a usize>,
1894 {
1895 let Some(parquet_index) = self.parquet_column_index else {
1896 let num_row_groups = row_group_indices.into_iter().count();
1897 return Ok(UInt64Array::new_null(num_row_groups));
1898 };
1899
1900 let iter = row_group_indices.into_iter().map(|rg_index| {
1901 let column_page_index_per_row_group_per_column =
1902 &column_page_index[*rg_index][parquet_index];
1903 let num_data_pages = &column_offset_index[*rg_index][parquet_index]
1904 .page_locations()
1905 .len();
1906
1907 (*num_data_pages, column_page_index_per_row_group_per_column)
1908 });
1909 null_counts_page_statistics(iter)
1910 }
1911
1912 pub fn data_page_row_counts<I>(
1930 &self,
1931 column_offset_index: &ParquetOffsetIndex,
1932 row_group_metadatas: &'a [RowGroupMetaData],
1933 row_group_indices: I,
1934 ) -> Result<Option<UInt64Array>>
1935 where
1936 I: IntoIterator<Item = &'a usize>,
1937 {
1938 let Some(parquet_index) = self.parquet_column_index else {
1939 return Ok(None);
1943 };
1944
1945 let mut row_counts = Vec::new();
1946 let mut nulls = NullBufferBuilder::new(0);
1947 for rg_idx in row_group_indices {
1948 let page_locations = &column_offset_index[*rg_idx][parquet_index].page_locations();
1949
1950 let row_count_per_page = page_locations
1951 .windows(2)
1952 .map(|loc| Some(loc[1].first_row_index as u64 - loc[0].first_row_index as u64));
1953
1954 let num_rows_in_row_group = &row_group_metadatas[*rg_idx].num_rows();
1956 let row_count_per_page = row_count_per_page.chain(std::iter::once(Some(
1957 *num_rows_in_row_group as u64
1958 - page_locations.last().unwrap().first_row_index as u64,
1959 )));
1960
1961 row_counts.extend(row_count_per_page.clone().map(|x| x.unwrap_or(0)));
1962 for val in row_count_per_page {
1963 if val.is_some() {
1964 nulls.append_non_null();
1965 } else {
1966 nulls.append_null();
1967 }
1968 }
1969 }
1970
1971 Ok(Some(UInt64Array::new(row_counts.into(), nulls.build())))
1972 }
1973
1974 fn make_null_array<I, A>(&self, data_type: &DataType, metadatas: I) -> ArrayRef
1976 where
1977 I: IntoIterator<Item = A>,
1978 {
1979 let num_row_groups = metadatas.into_iter().count();
1981 new_null_array(data_type, num_row_groups)
1982 }
1983}
1984
1985