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::RowGroupMetaData;
27use crate::file::metadata::page_index::PageIndexProvider;
28use crate::file::page_index::column_index::ColumnIndexMetaData;
29use crate::file::statistics::Statistics as ParquetStatistics;
30use crate::schema::types::SchemaDescriptor;
31use arrow_array::builder::{
32 BinaryBuilder, BinaryViewBuilder, BooleanBuilder, Date32Builder, Date64Builder,
33 Decimal32Builder, Decimal64Builder, FixedSizeBinaryBuilder, Float16Builder, Float32Builder,
34 Float64Builder, Int8Builder, Int16Builder, Int32Builder, Int64Builder, LargeBinaryBuilder,
35 LargeStringBuilder, StringBuilder, StringViewBuilder, Time32MillisecondBuilder,
36 Time32SecondBuilder, Time64MicrosecondBuilder, Time64NanosecondBuilder,
37 TimestampMicrosecondBuilder, TimestampMillisecondBuilder, TimestampNanosecondBuilder,
38 TimestampSecondBuilder, UInt8Builder, UInt16Builder, UInt32Builder, UInt64Builder,
39};
40use arrow_array::{
41 ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array, Decimal32Array, Decimal64Array,
42 Decimal128Array, Decimal256Array, Float16Array, Float32Array, Float64Array, Int8Array,
43 Int16Array, Int32Array, Int64Array, LargeBinaryArray, Time32MillisecondArray,
44 Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
45 TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt8Array,
46 UInt16Array, UInt32Array, UInt64Array, new_null_array,
47};
48use arrow_buffer::{NullBufferBuilder, i256};
49use arrow_schema::{DataType, Field, Schema, TimeUnit};
50use half::f16;
51use std::sync::Arc;
52
53pub(crate) fn from_bytes_to_i32(b: &[u8]) -> i32 {
56 i32::from_be_bytes(sign_extend_be::<4>(b))
60}
61
62pub(crate) fn from_bytes_to_i64(b: &[u8]) -> i64 {
65 i64::from_be_bytes(sign_extend_be::<8>(b))
66}
67
68pub(crate) fn from_bytes_to_i128(b: &[u8]) -> i128 {
71 i128::from_be_bytes(sign_extend_be::<16>(b))
72}
73
74pub(crate) fn from_bytes_to_i256(b: &[u8]) -> i256 {
77 i256::from_be_bytes(sign_extend_be::<32>(b))
78}
79
80pub(crate) fn from_bytes_to_f16(b: &[u8]) -> Option<f16> {
82 match b {
83 [low, high] => Some(f16::from_be_bytes([*high, *low])),
84 _ => None,
85 }
86}
87
88macro_rules! make_stats_iterator {
100 ($iterator_type:ident, $func:ident, $parquet_statistics_type:path, $stat_value_type:ty) => {
101 struct $iterator_type<'a, I>
108 where
109 I: Iterator<Item = Option<&'a ParquetStatistics>>,
110 {
111 iter: I,
112 }
113
114 impl<'a, I> $iterator_type<'a, I>
115 where
116 I: Iterator<Item = Option<&'a ParquetStatistics>>,
117 {
118 fn new(iter: I) -> Self {
120 Self { iter }
121 }
122 }
123
124 impl<'a, I> Iterator for $iterator_type<'a, I>
126 where
127 I: Iterator<Item = Option<&'a ParquetStatistics>>,
128 {
129 type Item = Option<&'a $stat_value_type>;
130
131 fn next(&mut self) -> Option<Self::Item> {
133 let next = self.iter.next();
134 next.map(|x| {
135 x.and_then(|stats| match stats {
136 $parquet_statistics_type(s) => s.$func(),
137 _ => None,
138 })
139 })
140 }
141
142 fn size_hint(&self) -> (usize, Option<usize>) {
143 self.iter.size_hint()
144 }
145 }
146 };
147}
148
149make_stats_iterator!(
150 MinBooleanStatsIterator,
151 min_opt,
152 ParquetStatistics::Boolean,
153 bool
154);
155make_stats_iterator!(
156 MaxBooleanStatsIterator,
157 max_opt,
158 ParquetStatistics::Boolean,
159 bool
160);
161make_stats_iterator!(
162 MinInt32StatsIterator,
163 min_opt,
164 ParquetStatistics::Int32,
165 i32
166);
167make_stats_iterator!(
168 MaxInt32StatsIterator,
169 max_opt,
170 ParquetStatistics::Int32,
171 i32
172);
173make_stats_iterator!(
174 MinInt64StatsIterator,
175 min_opt,
176 ParquetStatistics::Int64,
177 i64
178);
179make_stats_iterator!(
180 MaxInt64StatsIterator,
181 max_opt,
182 ParquetStatistics::Int64,
183 i64
184);
185make_stats_iterator!(
186 MinFloatStatsIterator,
187 min_opt,
188 ParquetStatistics::Float,
189 f32
190);
191make_stats_iterator!(
192 MaxFloatStatsIterator,
193 max_opt,
194 ParquetStatistics::Float,
195 f32
196);
197make_stats_iterator!(
198 MinDoubleStatsIterator,
199 min_opt,
200 ParquetStatistics::Double,
201 f64
202);
203make_stats_iterator!(
204 MaxDoubleStatsIterator,
205 max_opt,
206 ParquetStatistics::Double,
207 f64
208);
209make_stats_iterator!(
210 MinByteArrayStatsIterator,
211 min_bytes_opt,
212 ParquetStatistics::ByteArray,
213 [u8]
214);
215make_stats_iterator!(
216 MaxByteArrayStatsIterator,
217 max_bytes_opt,
218 ParquetStatistics::ByteArray,
219 [u8]
220);
221make_stats_iterator!(
222 MinFixedLenByteArrayStatsIterator,
223 min_bytes_opt,
224 ParquetStatistics::FixedLenByteArray,
225 [u8]
226);
227make_stats_iterator!(
228 MaxFixedLenByteArrayStatsIterator,
229 max_bytes_opt,
230 ParquetStatistics::FixedLenByteArray,
231 [u8]
232);
233
234macro_rules! make_decimal_stats_iterator {
252 ($iterator_type:ident, $func:ident, $bytes_func:ident, $stat_value_type:ident, $convert_func: ident) => {
253 struct $iterator_type<'a, I>
254 where
255 I: Iterator<Item = Option<&'a ParquetStatistics>>,
256 {
257 iter: I,
258 }
259
260 impl<'a, I> $iterator_type<'a, I>
261 where
262 I: Iterator<Item = Option<&'a ParquetStatistics>>,
263 {
264 fn new(iter: I) -> Self {
265 Self { iter }
266 }
267 }
268
269 impl<'a, I> Iterator for $iterator_type<'a, I>
270 where
271 I: Iterator<Item = Option<&'a ParquetStatistics>>,
272 {
273 type Item = Option<$stat_value_type>;
274
275 fn next(&mut self) -> Option<Self::Item> {
276 let next = self.iter.next();
277 next.map(|x| {
278 x.and_then(|stats| match stats {
279 ParquetStatistics::Int32(s) => {
280 s.$func().map(|x| $stat_value_type::from(*x))
281 }
282 ParquetStatistics::Int64(s) => s
283 .$func()
284 .map(|x| $stat_value_type::try_from(*x).ok())
285 .flatten(),
286 ParquetStatistics::ByteArray(s) => s.$bytes_func().map($convert_func),
287 ParquetStatistics::FixedLenByteArray(s) => {
288 s.$bytes_func().map($convert_func)
289 }
290 _ => None,
291 })
292 })
293 }
294
295 fn size_hint(&self) -> (usize, Option<usize>) {
296 self.iter.size_hint()
297 }
298 }
299 };
300}
301
302make_decimal_stats_iterator!(
303 MinDecimal32StatsIterator,
304 min_opt,
305 min_bytes_opt,
306 i32,
307 from_bytes_to_i32
308);
309make_decimal_stats_iterator!(
310 MaxDecimal32StatsIterator,
311 max_opt,
312 max_bytes_opt,
313 i32,
314 from_bytes_to_i32
315);
316make_decimal_stats_iterator!(
317 MinDecimal64StatsIterator,
318 min_opt,
319 min_bytes_opt,
320 i64,
321 from_bytes_to_i64
322);
323make_decimal_stats_iterator!(
324 MaxDecimal64StatsIterator,
325 max_opt,
326 max_bytes_opt,
327 i64,
328 from_bytes_to_i64
329);
330make_decimal_stats_iterator!(
331 MinDecimal128StatsIterator,
332 min_opt,
333 min_bytes_opt,
334 i128,
335 from_bytes_to_i128
336);
337make_decimal_stats_iterator!(
338 MaxDecimal128StatsIterator,
339 max_opt,
340 max_bytes_opt,
341 i128,
342 from_bytes_to_i128
343);
344make_decimal_stats_iterator!(
345 MinDecimal256StatsIterator,
346 min_opt,
347 min_bytes_opt,
348 i256,
349 from_bytes_to_i256
350);
351make_decimal_stats_iterator!(
352 MaxDecimal256StatsIterator,
353 max_opt,
354 max_bytes_opt,
355 i256,
356 from_bytes_to_i256
357);
358
359macro_rules! get_statistics {
366 (Min, $data_type: ident, $iterator: ident, $physical_type: ident) => {
367 get_statistics!(
368 $data_type,
369 $iterator,
370 $physical_type,
371 MinBooleanStatsIterator,
372 MinInt32StatsIterator,
373 MinInt64StatsIterator,
374 MinFloatStatsIterator,
375 MinDoubleStatsIterator,
376 MinByteArrayStatsIterator,
377 MinFixedLenByteArrayStatsIterator,
378 MinDecimal32StatsIterator,
379 MinDecimal64StatsIterator,
380 MinDecimal128StatsIterator,
381 MinDecimal256StatsIterator,
382 min_statistics
383 )
384 };
385 (Max, $data_type: ident, $iterator: ident, $physical_type: ident) => {
386 get_statistics!(
387 $data_type,
388 $iterator,
389 $physical_type,
390 MaxBooleanStatsIterator,
391 MaxInt32StatsIterator,
392 MaxInt64StatsIterator,
393 MaxFloatStatsIterator,
394 MaxDoubleStatsIterator,
395 MaxByteArrayStatsIterator,
396 MaxFixedLenByteArrayStatsIterator,
397 MaxDecimal32StatsIterator,
398 MaxDecimal64StatsIterator,
399 MaxDecimal128StatsIterator,
400 MaxDecimal256StatsIterator,
401 max_statistics
402 )
403 };
404 (
405 $data_type: ident,
406 $iterator: ident,
407 $physical_type: ident,
408 $boolean_iter: ident,
409 $int32_iter: ident,
410 $int64_iter: ident,
411 $float_iter: ident,
412 $double_iter: ident,
413 $byte_array_iter: ident,
414 $fixed_len_byte_array_iter: ident,
415 $decimal32_iter: ident,
416 $decimal64_iter: ident,
417 $decimal128_iter: ident,
418 $decimal256_iter: ident,
419 $dictionary_statistics: ident
420 ) => {
421 match $data_type {
422 DataType::Boolean => Ok(Arc::new(BooleanArray::from_iter(
423 $boolean_iter::new($iterator).map(|x| x.copied()),
424 ))),
425 DataType::Int8 => Ok(Arc::new(Int8Array::from_iter(
426 $int32_iter::new($iterator).map(|x| {
427 x.and_then(|x| i8::try_from(*x).ok())
428 }),
429 ))),
430 DataType::Int16 => Ok(Arc::new(Int16Array::from_iter(
431 $int32_iter::new($iterator).map(|x| {
432 x.and_then(|x| i16::try_from(*x).ok())
433 }),
434 ))),
435 DataType::Int32 => Ok(Arc::new(Int32Array::from_iter(
436 $int32_iter::new($iterator).map(|x| x.copied()),
437 ))),
438 DataType::Int64 => Ok(Arc::new(Int64Array::from_iter(
439 $int64_iter::new($iterator).map(|x| x.copied()),
440 ))),
441 DataType::UInt8 => Ok(Arc::new(UInt8Array::from_iter(
442 $int32_iter::new($iterator).map(|x| {
443 x.and_then(|x| u8::try_from(*x).ok())
444 }),
445 ))),
446 DataType::UInt16 => Ok(Arc::new(UInt16Array::from_iter(
447 $int32_iter::new($iterator).map(|x| {
448 x.and_then(|x| u16::try_from(*x).ok())
449 }),
450 ))),
451 DataType::UInt32 => Ok(Arc::new(UInt32Array::from_iter(
452 $int32_iter::new($iterator).map(|x| x.map(|x| *x as u32)),
453 ))),
454 DataType::UInt64 => Ok(Arc::new(UInt64Array::from_iter(
455 $int64_iter::new($iterator).map(|x| x.map(|x| *x as u64)),
456 ))),
457 DataType::Float16 => Ok(Arc::new(Float16Array::from_iter(
458 $fixed_len_byte_array_iter::new($iterator).map(|x| x.and_then(|x| {
459 from_bytes_to_f16(x)
460 })),
461 ))),
462 DataType::Float32 => Ok(Arc::new(Float32Array::from_iter(
463 $float_iter::new($iterator).map(|x| x.copied()),
464 ))),
465 DataType::Float64 => Ok(Arc::new(Float64Array::from_iter(
466 $double_iter::new($iterator).map(|x| x.copied()),
467 ))),
468 DataType::Date32 => Ok(Arc::new(Date32Array::from_iter(
469 $int32_iter::new($iterator).map(|x| x.copied()),
470 ))),
471 DataType::Date64 if $physical_type == Some(PhysicalType::INT32) => Ok(Arc::new(Date64Array::from_iter(
472 $int32_iter::new($iterator)
473 .map(|x| x.map(|x| i64::from(*x) * 24 * 60 * 60 * 1000))))),
474 DataType::Date64 if $physical_type == Some(PhysicalType::INT64) => Ok(Arc::new(Date64Array::from_iter(
475 $int64_iter::new($iterator).map(|x| x.copied()),))),
476 DataType::Timestamp(unit, timezone) =>{
477 let iter = $int64_iter::new($iterator).map(|x| x.copied());
478 Ok(match unit {
479 TimeUnit::Second => Arc::new(TimestampSecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
480 TimeUnit::Millisecond => Arc::new(TimestampMillisecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
481 TimeUnit::Microsecond => Arc::new(TimestampMicrosecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
482 TimeUnit::Nanosecond => Arc::new(TimestampNanosecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
483 })
484 },
485 DataType::Time32(unit) => {
486 Ok(match unit {
487 TimeUnit::Second => Arc::new(Time32SecondArray::from_iter(
488 $int32_iter::new($iterator).map(|x| x.copied()),
489 )),
490 TimeUnit::Millisecond => Arc::new(Time32MillisecondArray::from_iter(
491 $int32_iter::new($iterator).map(|x| x.copied()),
492 )),
493 _ => {
494 let len = $iterator.count();
495 new_null_array($data_type, len)
497 }
498 })
499 },
500 DataType::Time64(unit) => {
501 Ok(match unit {
502 TimeUnit::Microsecond => Arc::new(Time64MicrosecondArray::from_iter(
503 $int64_iter::new($iterator).map(|x| x.copied()),
504 )),
505 TimeUnit::Nanosecond => Arc::new(Time64NanosecondArray::from_iter(
506 $int64_iter::new($iterator).map(|x| x.copied()),
507 )),
508 _ => {
509 let len = $iterator.count();
510 new_null_array($data_type, len)
512 }
513 })
514 },
515 DataType::Binary => Ok(Arc::new(BinaryArray::from_iter(
516 $byte_array_iter::new($iterator)
517 ))),
518 DataType::LargeBinary => Ok(Arc::new(LargeBinaryArray::from_iter(
519 $byte_array_iter::new($iterator)
520 ))),
521 DataType::Utf8 => {
522 let iterator = $byte_array_iter::new($iterator);
523 let mut builder = StringBuilder::new();
524 for x in iterator {
525 let Some(x) = x else {
526 builder.append_null(); continue;
528 };
529
530 let Ok(x) = std::str::from_utf8(x) else {
531 builder.append_null();
532 continue;
533 };
534
535 builder.append_value(x);
536 }
537 Ok(Arc::new(builder.finish()))
538 },
539 DataType::LargeUtf8 => {
540 let iterator = $byte_array_iter::new($iterator);
541 let mut builder = LargeStringBuilder::new();
542 for x in iterator {
543 let Some(x) = x else {
544 builder.append_null(); continue;
546 };
547
548 let Ok(x) = std::str::from_utf8(x) else {
549 builder.append_null();
550 continue;
551 };
552
553 builder.append_value(x);
554 }
555 Ok(Arc::new(builder.finish()))
556 },
557 DataType::FixedSizeBinary(size) => {
558 let iterator = $fixed_len_byte_array_iter::new($iterator);
559 let mut builder = FixedSizeBinaryBuilder::new(*size);
560 for x in iterator {
561 let Some(x) = x else {
562 builder.append_null(); continue;
564 };
565
566 if x.len().try_into() != Ok(*size){
568 builder.append_null();
569 continue;
570 }
571
572 builder.append_value(x).expect("ensure to append successfully here, because size have been checked before");
573 }
574 Ok(Arc::new(builder.finish()))
575 },
576 DataType::Decimal32(precision, scale) => {
577 let arr = Decimal32Array::from_iter(
578 $decimal32_iter::new($iterator)
579 ).with_precision_and_scale(*precision, *scale)?;
580 Ok(Arc::new(arr))
581 },
582 DataType::Decimal64(precision, scale) => {
583 let arr = Decimal64Array::from_iter(
584 $decimal64_iter::new($iterator)
585 ).with_precision_and_scale(*precision, *scale)?;
586 Ok(Arc::new(arr))
587 },
588 DataType::Decimal128(precision, scale) => {
589 let arr = Decimal128Array::from_iter(
590 $decimal128_iter::new($iterator)
591 ).with_precision_and_scale(*precision, *scale)?;
592 Ok(Arc::new(arr))
593 },
594 DataType::Decimal256(precision, scale) => {
595 let arr = Decimal256Array::from_iter(
596 $decimal256_iter::new($iterator)
597 ).with_precision_and_scale(*precision, *scale)?;
598 Ok(Arc::new(arr))
599 },
600 DataType::Dictionary(_, value_type) => {
601 $dictionary_statistics(value_type, $iterator, $physical_type)
602 },
603 DataType::Utf8View => {
604 let iterator = $byte_array_iter::new($iterator);
605 let mut builder = StringViewBuilder::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::BinaryView => {
622 let iterator = $byte_array_iter::new($iterator);
623 let mut builder = BinaryViewBuilder::new();
624 for x in iterator {
625 let Some(x) = x else {
626 builder.append_null(); continue;
628 };
629
630 builder.append_value(x);
631 }
632 Ok(Arc::new(builder.finish()))
633 }
634
635 DataType::Map(_,_) |
636 DataType::Duration(_) |
637 DataType::Interval(_) |
638 DataType::Date64 | DataType::Null |
640 DataType::List(_) |
641 DataType::ListView(_) |
642 DataType::FixedSizeList(_, _) |
643 DataType::LargeList(_) |
644 DataType::LargeListView(_) |
645 DataType::Struct(_) |
646 DataType::Union(_, _) |
647 DataType::RunEndEncoded(_, _) => {
648 let len = $iterator.count();
649 Ok(new_null_array($data_type, len))
651 }
652 }
653 };
654}
655
656macro_rules! get_data_page_statistics {
657 (Min, $data_type: ident, $iterator: ident, $physical_type: ident) => {
658 get_data_page_statistics!(
659 $data_type,
660 $iterator,
661 $physical_type,
662 min_values_iter,
663 min_page_statistics
664 )
665 };
666 (Max, $data_type: ident, $iterator: ident, $physical_type: ident) => {
667 get_data_page_statistics!(
668 $data_type,
669 $iterator,
670 $physical_type,
671 max_values_iter,
672 max_page_statistics
673 )
674 };
675 (
676 $data_type: ident,
677 $iterator: ident,
678 $physical_type: ident,
679 $values_iter: ident,
680 $page_statistics: ident
681 ) => {{
682 let chunks: Vec<(usize, Option<&ColumnIndexMetaData>)> = $iterator.collect();
683 let capacity: usize = chunks.iter().map(|c| c.0).sum();
684 match $data_type {
685 DataType::Boolean => {
686 let mut b = BooleanBuilder::with_capacity(capacity);
687 for (len, index) in chunks {
688 match index {
689 Some(ColumnIndexMetaData::BOOLEAN(index)) => {
690 for val in index.$values_iter() {
691 b.append_option(val.copied());
692 }
693 }
694 _ => b.append_nulls(len),
695 }
696 }
697 Ok(Arc::new(b.finish()))
698 },
699 DataType::UInt8 => {
700 let mut b = UInt8Builder::with_capacity(capacity);
701 for (len, index) in chunks {
702 match index {
703 Some(ColumnIndexMetaData::INT32(index)) => {
704 b.extend_from_iter_option(
705 index.$values_iter()
706 .map(|val| val.and_then(|&x| u8::try_from(x).ok())),
707 );
708 }
709 _ => b.append_nulls(len),
710 }
711 }
712 Ok(Arc::new(b.finish()))
713 },
714 DataType::UInt16 => {
715 let mut b = UInt16Builder::with_capacity(capacity);
716 for (len, index) in chunks {
717 match index {
718 Some(ColumnIndexMetaData::INT32(index)) => {
719 b.extend_from_iter_option(
720 index.$values_iter()
721 .map(|val| val.and_then(|&x| u16::try_from(x).ok())),
722 );
723 }
724 _ => b.append_nulls(len),
725 }
726 }
727 Ok(Arc::new(b.finish()))
728 },
729 DataType::UInt32 => {
730 let mut b = UInt32Builder::with_capacity(capacity);
731 for (len, index) in chunks {
732 match index {
733 Some(ColumnIndexMetaData::INT32(index)) => {
734 b.extend_from_iter_option(
735 index.$values_iter()
736 .map(|val| val.map(|&x| x as u32)),
737 );
738 }
739 _ => b.append_nulls(len),
740 }
741 }
742 Ok(Arc::new(b.finish()))
743 },
744 DataType::UInt64 => {
745 let mut b = UInt64Builder::with_capacity(capacity);
746 for (len, index) in chunks {
747 match index {
748 Some(ColumnIndexMetaData::INT64(index)) => {
749 b.extend_from_iter_option(
750 index.$values_iter()
751 .map(|val| val.map(|&x| x as u64)),
752 );
753 }
754 _ => b.append_nulls(len),
755 }
756 }
757 Ok(Arc::new(b.finish()))
758 },
759 DataType::Int8 => {
760 let mut b = Int8Builder::with_capacity(capacity);
761 for (len, index) in chunks {
762 match index {
763 Some(ColumnIndexMetaData::INT32(index)) => {
764 b.extend_from_iter_option(
765 index.$values_iter()
766 .map(|val| val.and_then(|&x| i8::try_from(x).ok())),
767 );
768 }
769 _ => b.append_nulls(len),
770 }
771 }
772 Ok(Arc::new(b.finish()))
773 },
774 DataType::Int16 => {
775 let mut b = Int16Builder::with_capacity(capacity);
776 for (len, index) in chunks {
777 match index {
778 Some(ColumnIndexMetaData::INT32(index)) => {
779 b.extend_from_iter_option(
780 index.$values_iter()
781 .map(|val| val.and_then(|&x| i16::try_from(x).ok())),
782 );
783 }
784 _ => b.append_nulls(len),
785 }
786 }
787 Ok(Arc::new(b.finish()))
788 },
789 DataType::Int32 => {
790 let mut b = Int32Builder::with_capacity(capacity);
791 for (len, index) in chunks {
792 match index {
793 Some(ColumnIndexMetaData::INT32(index)) => {
794 b.extend_from_iter_option(
795 index.$values_iter()
796 .map(|val| val.copied()),
797 );
798 }
799 _ => b.append_nulls(len),
800 }
801 }
802 Ok(Arc::new(b.finish()))
803 },
804 DataType::Int64 => {
805 let mut b = Int64Builder::with_capacity(capacity);
806 for (len, index) in chunks {
807 match index {
808 Some(ColumnIndexMetaData::INT64(index)) => {
809 b.extend_from_iter_option(
810 index.$values_iter()
811 .map(|val| val.copied()),
812 );
813 }
814 _ => b.append_nulls(len),
815 }
816 }
817 Ok(Arc::new(b.finish()))
818 },
819 DataType::Float16 => {
820 let mut b = Float16Builder::with_capacity(capacity);
821 for (len, index) in chunks {
822 match index {
823 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
824 b.extend_from_iter_option(
825 index.$values_iter()
826 .map(|val| val.and_then(|x| from_bytes_to_f16(x))),
827 );
828 }
829 _ => b.append_nulls(len),
830 }
831 }
832 Ok(Arc::new(b.finish()))
833 },
834 DataType::Float32 => {
835 let mut b = Float32Builder::with_capacity(capacity);
836 for (len, index) in chunks {
837 match index {
838 Some(ColumnIndexMetaData::FLOAT(index)) => {
839 b.extend_from_iter_option(
840 index.$values_iter()
841 .map(|val| val.copied()),
842 );
843 }
844 _ => b.append_nulls(len),
845 }
846 }
847 Ok(Arc::new(b.finish()))
848 },
849 DataType::Float64 => {
850 let mut b = Float64Builder::with_capacity(capacity);
851 for (len, index) in chunks {
852 match index {
853 Some(ColumnIndexMetaData::DOUBLE(index)) => {
854 b.extend_from_iter_option(
855 index.$values_iter()
856 .map(|val| val.copied()),
857 );
858 }
859 _ => b.append_nulls(len),
860 }
861 }
862 Ok(Arc::new(b.finish()))
863 },
864 DataType::Binary => {
865 let mut b = BinaryBuilder::with_capacity(capacity, capacity * 10);
866 for (len, index) in chunks {
867 match index {
868 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
869 for val in index.$values_iter() {
870 b.append_option(val.map(|x| x.as_ref()));
871 }
872 }
873 _ => b.append_nulls(len),
874 }
875 }
876 Ok(Arc::new(b.finish()))
877 },
878 DataType::LargeBinary => {
879 let mut b = LargeBinaryBuilder::with_capacity(capacity, capacity * 10);
880 for (len, index) in chunks {
881 match index {
882 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
883 for val in index.$values_iter() {
884 b.append_option(val.map(|x| x.as_ref()));
885 }
886 }
887 _ => b.append_nulls(len),
888 }
889 }
890 Ok(Arc::new(b.finish()))
891 },
892 DataType::Utf8 => {
893 let mut b = StringBuilder::with_capacity(capacity, capacity * 10);
894 for (len, index) in chunks {
895 match index {
896 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
897 for val in index.$values_iter() {
898 match val {
899 Some(x) => match std::str::from_utf8(x.as_ref()) {
900 Ok(s) => b.append_value(s),
901 _ => b.append_null(),
902 }
903 None => b.append_null(),
904 }
905 }
906 }
907 _ => b.append_nulls(len),
908 }
909 }
910 Ok(Arc::new(b.finish()))
911 },
912 DataType::LargeUtf8 => {
913 let mut b = LargeStringBuilder::with_capacity(capacity, capacity * 10);
914 for (len, index) in chunks {
915 match index {
916 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
917 for val in index.$values_iter() {
918 match val {
919 Some(x) => match std::str::from_utf8(x.as_ref()) {
920 Ok(s) => b.append_value(s),
921 _ => b.append_null(),
922 }
923 None => b.append_null(),
924 }
925 }
926 }
927 _ => b.append_nulls(len),
928 }
929 }
930 Ok(Arc::new(b.finish()))
931 },
932 DataType::Dictionary(_, value_type) => {
933 $page_statistics(value_type, chunks.into_iter(), $physical_type)
934 },
935 DataType::Timestamp(unit, timezone) => {
936 match unit {
937 TimeUnit::Second => {
938 let mut b = TimestampSecondBuilder::with_capacity(capacity);
939 for (len, index) in chunks {
940 match index {
941 Some(ColumnIndexMetaData::INT64(index)) => {
942 b.extend_from_iter_option(
943 index.$values_iter()
944 .map(|val| val.copied()),
945 );
946 }
947 _ => b.append_nulls(len),
948 }
949 }
950 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
951 }
952 TimeUnit::Millisecond => {
953 let mut b = TimestampMillisecondBuilder::with_capacity(capacity);
954 for (len, index) in chunks {
955 match index {
956 Some(ColumnIndexMetaData::INT64(index)) => {
957 b.extend_from_iter_option(
958 index.$values_iter()
959 .map(|val| val.copied()),
960 );
961 }
962 _ => b.append_nulls(len),
963 }
964 }
965 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
966 }
967 TimeUnit::Microsecond => {
968 let mut b = TimestampMicrosecondBuilder::with_capacity(capacity);
969 for (len, index) in chunks {
970 match index {
971 Some(ColumnIndexMetaData::INT64(index)) => {
972 b.extend_from_iter_option(
973 index.$values_iter()
974 .map(|val| val.copied()),
975 );
976 }
977 _ => b.append_nulls(len),
978 }
979 }
980 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
981 }
982 TimeUnit::Nanosecond => {
983 let mut b = TimestampNanosecondBuilder::with_capacity(capacity);
984 for (len, index) in chunks {
985 match index {
986 Some(ColumnIndexMetaData::INT64(index)) => {
987 b.extend_from_iter_option(
988 index.$values_iter()
989 .map(|val| val.copied()),
990 );
991 }
992 _ => b.append_nulls(len),
993 }
994 }
995 Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
996 }
997 }
998 },
999 DataType::Date32 => {
1000 let mut b = Date32Builder::with_capacity(capacity);
1001 for (len, index) in chunks {
1002 match index {
1003 Some(ColumnIndexMetaData::INT32(index)) => {
1004 b.extend_from_iter_option(
1005 index.$values_iter()
1006 .map(|val| val.copied()),
1007 );
1008 }
1009 _ => b.append_nulls(len),
1010 }
1011 }
1012 Ok(Arc::new(b.finish()))
1013 },
1014 DataType::Date64 if $physical_type == Some(PhysicalType::INT32)=> {
1015 let mut b = Date64Builder::with_capacity(capacity);
1016 for (len, index) in chunks {
1017 match index {
1018 Some(ColumnIndexMetaData::INT32(index)) => {
1019 b.extend_from_iter_option(
1020 index.$values_iter()
1021 .map(|val| val.map(|&x| (x as i64) * 24 * 60 * 60 * 1000)),
1022 );
1023 }
1024 _ => b.append_nulls(len),
1025 }
1026 }
1027 Ok(Arc::new(b.finish()))
1028 },
1029 DataType::Date64 if $physical_type == Some(PhysicalType::INT64) => {
1030 let mut b = Date64Builder::with_capacity(capacity);
1031 for (len, index) in chunks {
1032 match index {
1033 Some(ColumnIndexMetaData::INT64(index)) => {
1034 b.extend_from_iter_option(
1035 index.$values_iter()
1036 .map(|val| val.copied()),
1037 );
1038 }
1039 _ => b.append_nulls(len),
1040 }
1041 }
1042 Ok(Arc::new(b.finish()))
1043 },
1044 DataType::Decimal32(precision, scale) => {
1045 let mut b = Decimal32Builder::with_capacity(capacity);
1046 for (len, index) in chunks {
1047 match index {
1048 Some(ColumnIndexMetaData::INT32(index)) => {
1049 b.extend_from_iter_option(
1050 index.$values_iter()
1051 .map(|val| val.copied()),
1052 );
1053 }
1054 Some(ColumnIndexMetaData::INT64(index)) => {
1055 b.extend_from_iter_option(
1056 index.$values_iter()
1057 .map(|val| val.and_then(|&x| i32::try_from(x).ok())),
1058 );
1059 }
1060 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1061 b.extend_from_iter_option(
1062 index.$values_iter()
1063 .map(|val| val.map(|x| from_bytes_to_i32(x.as_ref()))),
1064 );
1065 }
1066 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1067 b.extend_from_iter_option(
1068 index.$values_iter()
1069 .map(|val| val.map(|x| from_bytes_to_i32(x.as_ref()))),
1070 );
1071 }
1072 _ => b.append_nulls(len),
1073 }
1074 }
1075 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1076 },
1077 DataType::Decimal64(precision, scale) => {
1078 let mut b = Decimal64Builder::with_capacity(capacity);
1079 for (len, index) in chunks {
1080 match index {
1081 Some(ColumnIndexMetaData::INT32(index)) => {
1082 b.extend_from_iter_option(
1083 index.$values_iter()
1084 .map(|val| val.map(|x| *x as i64)),
1085 );
1086 }
1087 Some(ColumnIndexMetaData::INT64(index)) => {
1088 b.extend_from_iter_option(
1089 index.$values_iter()
1090 .map(|val| val.copied()),
1091 );
1092 }
1093 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1094 b.extend_from_iter_option(
1095 index.$values_iter()
1096 .map(|val| val.map(|x| from_bytes_to_i64(x.as_ref()))),
1097 );
1098 }
1099 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1100 b.extend_from_iter_option(
1101 index.$values_iter()
1102 .map(|val| val.map(|x| from_bytes_to_i64(x.as_ref()))),
1103 );
1104 }
1105 _ => b.append_nulls(len),
1106 }
1107 }
1108 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1109 },
1110 DataType::Decimal128(precision, scale) => {
1111 let mut b = Decimal128Array::builder(capacity);
1112 for (len, index) in chunks {
1113 match index {
1114 Some(ColumnIndexMetaData::INT32(index)) => {
1115 b.extend_from_iter_option(
1116 index.$values_iter()
1117 .map(|val| val.map(|x| *x as i128)),
1118 );
1119 }
1120 Some(ColumnIndexMetaData::INT64(index)) => {
1121 b.extend_from_iter_option(
1122 index.$values_iter()
1123 .map(|val| val.map(|x| *x as i128)),
1124 );
1125 }
1126 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1127 b.extend_from_iter_option(
1128 index.$values_iter()
1129 .map(|val| val.map(|x| from_bytes_to_i128(x.as_ref()))),
1130 );
1131 }
1132 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1133 b.extend_from_iter_option(
1134 index.$values_iter()
1135 .map(|val| val.map(|x| from_bytes_to_i128(x.as_ref()))),
1136 );
1137 }
1138 _ => b.append_nulls(len),
1139 }
1140 }
1141 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1142 },
1143 DataType::Decimal256(precision, scale) => {
1144 let mut b = Decimal256Array::builder(capacity);
1145 for (len, index) in chunks {
1146 match index {
1147 Some(ColumnIndexMetaData::INT32(index)) => {
1148 b.extend_from_iter_option(
1149 index.$values_iter()
1150 .map(|val| val.map(|x| i256::from_i128(*x as i128))),
1151 );
1152 }
1153 Some(ColumnIndexMetaData::INT64(index)) => {
1154 b.extend_from_iter_option(
1155 index.$values_iter()
1156 .map(|val| val.map(|x| i256::from_i128(*x as i128))),
1157 );
1158 }
1159 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1160 b.extend_from_iter_option(
1161 index.$values_iter()
1162 .map(|val| val.map(|x| from_bytes_to_i256(x.as_ref()))),
1163 );
1164 }
1165 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1166 b.extend_from_iter_option(
1167 index.$values_iter()
1168 .map(|val| val.map(|x| from_bytes_to_i256(x.as_ref()))),
1169 );
1170 }
1171 _ => b.append_nulls(len),
1172 }
1173 }
1174 Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
1175 },
1176 DataType::Time32(unit) => {
1177 match unit {
1178 TimeUnit::Second => {
1179 let mut b = Time32SecondBuilder::with_capacity(capacity);
1180 for (len, index) in chunks {
1181 match index {
1182 Some(ColumnIndexMetaData::INT32(index)) => {
1183 b.extend_from_iter_option(
1184 index.$values_iter()
1185 .map(|val| val.copied()),
1186 );
1187 }
1188 _ => b.append_nulls(len),
1189 }
1190 }
1191 Ok(Arc::new(b.finish()))
1192 }
1193 TimeUnit::Millisecond => {
1194 let mut b = Time32MillisecondBuilder::with_capacity(capacity);
1195 for (len, index) in chunks {
1196 match index {
1197 Some(ColumnIndexMetaData::INT32(index)) => {
1198 b.extend_from_iter_option(
1199 index.$values_iter()
1200 .map(|val| val.copied()),
1201 );
1202 }
1203 _ => b.append_nulls(len),
1204 }
1205 }
1206 Ok(Arc::new(b.finish()))
1207 }
1208 _ => {
1209 Ok(new_null_array($data_type, capacity))
1210 }
1211 }
1212 }
1213 DataType::Time64(unit) => {
1214 match unit {
1215 TimeUnit::Microsecond => {
1216 let mut b = Time64MicrosecondBuilder::with_capacity(capacity);
1217 for (len, index) in chunks {
1218 match index {
1219 Some(ColumnIndexMetaData::INT64(index)) => {
1220 b.extend_from_iter_option(
1221 index.$values_iter()
1222 .map(|val| val.copied()),
1223 );
1224 }
1225 _ => b.append_nulls(len),
1226 }
1227 }
1228 Ok(Arc::new(b.finish()))
1229 }
1230 TimeUnit::Nanosecond => {
1231 let mut b = Time64NanosecondBuilder::with_capacity(capacity);
1232 for (len, index) in chunks {
1233 match index {
1234 Some(ColumnIndexMetaData::INT64(index)) => {
1235 b.extend_from_iter_option(
1236 index.$values_iter()
1237 .map(|val| val.copied()),
1238 );
1239 }
1240 _ => b.append_nulls(len),
1241 }
1242 }
1243 Ok(Arc::new(b.finish()))
1244 }
1245 _ => {
1246 Ok(new_null_array($data_type, capacity))
1247 }
1248 }
1249 },
1250 DataType::FixedSizeBinary(size) => {
1251 let mut b = FixedSizeBinaryBuilder::with_capacity(capacity, *size);
1252 for (len, index) in chunks {
1253 match index {
1254 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
1255 for val in index.$values_iter() {
1256 match val {
1257 Some(v) => {
1258 if v.len() == *size as usize {
1259 let _ = b.append_value(v.as_ref())?;
1260 } else {
1261 b.append_null();
1262 }
1263 }
1264 None => b.append_null(),
1265 }
1266 }
1267 }
1268 _ => b.append_nulls(len),
1269 }
1270 }
1271 Ok(Arc::new(b.finish()))
1272 },
1273 DataType::Utf8View => {
1274 let mut b = StringViewBuilder::with_capacity(capacity);
1275 for (len, index) in chunks {
1276 match index {
1277 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1278 for val in index.$values_iter() {
1279 match val {
1280 Some(x) => match std::str::from_utf8(x.as_ref()) {
1281 Ok(s) => b.append_value(s),
1282 _ => b.append_null(),
1283 }
1284 None => b.append_null(),
1285 }
1286 }
1287 }
1288 _ => {
1289 for _ in 0..len { b.append_null(); }
1290 }
1291 }
1292 }
1293 Ok(Arc::new(b.finish()))
1294 },
1295 DataType::BinaryView => {
1296 let mut b = BinaryViewBuilder::with_capacity(capacity);
1297 for (len, index) in chunks {
1298 match index {
1299 Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
1300 for val in index.$values_iter() {
1301 match val {
1302 Some(v) => b.append_value(v.as_ref()),
1303 None => b.append_null(),
1304 }
1305 }
1306 }
1307 _ => {
1308 for _ in 0..len { b.append_null(); }
1309 }
1310 }
1311 }
1312 Ok(Arc::new(b.finish()))
1313 },
1314 DataType::Date64 | DataType::Null |
1316 DataType::Duration(_) |
1317 DataType::Interval(_) |
1318 DataType::List(_) |
1319 DataType::ListView(_) |
1320 DataType::FixedSizeList(_, _) |
1321 DataType::LargeList(_) |
1322 DataType::LargeListView(_) |
1323 DataType::Struct(_) |
1324 DataType::Union(_, _) |
1325 DataType::Map(_, _) |
1326 DataType::RunEndEncoded(_, _) => {
1327 Ok(new_null_array($data_type, capacity))
1329 },
1330 }
1331 }
1332 };
1333}
1334fn min_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
1339 data_type: &DataType,
1340 iterator: I,
1341 physical_type: Option<PhysicalType>,
1342) -> Result<ArrayRef> {
1343 get_statistics!(Min, data_type, iterator, physical_type)
1344}
1345
1346fn max_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
1350 data_type: &DataType,
1351 iterator: I,
1352 physical_type: Option<PhysicalType>,
1353) -> Result<ArrayRef> {
1354 get_statistics!(Max, data_type, iterator, physical_type)
1355}
1356
1357pub(crate) fn min_page_statistics<'a, I>(
1360 data_type: &DataType,
1361 iterator: I,
1362 physical_type: Option<PhysicalType>,
1363) -> Result<ArrayRef>
1364where
1365 I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
1366{
1367 get_data_page_statistics!(Min, data_type, iterator, physical_type)
1368}
1369
1370pub(crate) fn max_page_statistics<'a, I>(
1373 data_type: &DataType,
1374 iterator: I,
1375 physical_type: Option<PhysicalType>,
1376) -> Result<ArrayRef>
1377where
1378 I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
1379{
1380 get_data_page_statistics!(Max, data_type, iterator, physical_type)
1381}
1382
1383pub(crate) fn null_counts_page_statistics<'a, I>(iterator: I) -> Result<UInt64Array>
1388where
1389 I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
1390{
1391 let chunks: Vec<_> = iterator.collect();
1392 let total_capacity: usize = chunks.iter().map(|(len, _)| *len).sum();
1393 let mut values = Vec::with_capacity(total_capacity);
1394 let mut nulls = NullBufferBuilder::new(total_capacity);
1395 for (len, index) in chunks {
1396 match index {
1397 Some(index) if index.null_counts().is_some() => {
1398 values.extend(index.null_counts().unwrap().iter().map(|&x| x as u64));
1399 nulls.append_n_non_nulls(len);
1400 }
1401 _ => {
1402 values.resize(values.len() + len, 0);
1403 nulls.append_n_nulls(len);
1404 }
1405 }
1406 }
1407 let null_buffer = nulls.build();
1408 let array = UInt64Array::new(values.into(), null_buffer);
1409 Ok(array)
1410}
1411
1412pub(crate) fn nan_counts_page_statistics<'a, I>(iterator: I) -> Result<UInt64Array>
1417where
1418 I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
1419{
1420 let chunks: Vec<_> = iterator.collect();
1421 let total_capacity: usize = chunks.iter().map(|(len, _)| *len).sum();
1422 let mut values = Vec::with_capacity(total_capacity);
1423 let mut nulls = NullBufferBuilder::new(total_capacity);
1424 for (len, index) in chunks {
1425 match index {
1426 Some(index) if index.nan_counts().is_some() => {
1427 values.extend(index.nan_counts().unwrap().iter().map(|&x| x as u64));
1428 nulls.append_n_non_nulls(len);
1429 }
1430 _ => {
1431 values.resize(values.len() + len, 0);
1432 nulls.append_n_nulls(len);
1433 }
1434 }
1435 }
1436 let null_buffer = nulls.build();
1437 let array = UInt64Array::new(values.into(), null_buffer);
1438 Ok(array)
1439}
1440
1441#[derive(Debug)]
1464pub struct StatisticsConverter<'a> {
1465 parquet_column_index: Option<usize>,
1467 arrow_field: &'a Field,
1469 missing_null_counts_as_zero: bool,
1471 physical_type: Option<PhysicalType>,
1473}
1474
1475impl<'a> StatisticsConverter<'a> {
1476 pub fn parquet_column_index(&self) -> Option<usize> {
1481 self.parquet_column_index
1482 }
1483
1484 pub fn arrow_field(&self) -> &'a Field {
1486 self.arrow_field
1487 }
1488
1489 pub fn with_missing_null_counts_as_zero(mut self, missing_null_counts_as_zero: bool) -> Self {
1502 self.missing_null_counts_as_zero = missing_null_counts_as_zero;
1503 self
1504 }
1505
1506 pub fn row_group_row_counts<I>(&self, metadatas: I) -> Result<Option<UInt64Array>>
1537 where
1538 I: IntoIterator<Item = &'a RowGroupMetaData>,
1539 {
1540 let Some(_) = self.parquet_column_index else {
1541 return Ok(None);
1542 };
1543
1544 let mut builder = UInt64Array::builder(10);
1545 for metadata in metadatas {
1546 let row_count = metadata.num_rows();
1547 let row_count: u64 = row_count.try_into().map_err(|e| {
1548 arrow_err!(format!(
1549 "Parquet row count {row_count} too large to convert to u64: {e}"
1550 ))
1551 })?;
1552 builder.append_value(row_count);
1553 }
1554 Ok(Some(builder.finish()))
1555 }
1556
1557 pub fn try_new<'b>(
1572 column_name: &'b str,
1573 arrow_schema: &'a Schema,
1574 parquet_schema: &'a SchemaDescriptor,
1575 ) -> Result<Self> {
1576 let Some((_idx, arrow_field)) = arrow_schema.column_with_name(column_name) else {
1578 return Err(arrow_err!(format!(
1579 "Column '{}' not found in schema for statistics conversion",
1580 column_name
1581 )));
1582 };
1583
1584 let parquet_index = match parquet_column(parquet_schema, arrow_schema, column_name) {
1586 Some((parquet_idx, matched_field)) => {
1587 if matched_field.as_ref() != arrow_field {
1589 return Err(arrow_err!(format!(
1590 "Matched column '{:?}' does not match original matched column '{:?}'",
1591 matched_field, arrow_field
1592 )));
1593 }
1594 Some(parquet_idx)
1595 }
1596 None => None,
1597 };
1598
1599 Ok(Self {
1600 parquet_column_index: parquet_index,
1601 arrow_field,
1602 missing_null_counts_as_zero: true,
1603 physical_type: parquet_index.map(|idx| parquet_schema.column(idx).physical_type()),
1604 })
1605 }
1606
1607 pub fn from_column_index(
1629 parquet_column_index: usize,
1630 arrow_field: &'a Field,
1631 parquet_schema: &'a SchemaDescriptor,
1632 ) -> Result<Self> {
1633 if parquet_column_index >= parquet_schema.columns().len() {
1634 return Err(arrow_err!(format!(
1635 "Parquet column index {} out of bounds, column count {}",
1636 parquet_column_index,
1637 parquet_schema.columns().len()
1638 )));
1639 }
1640
1641 let physical_type = parquet_schema.column(parquet_column_index).physical_type();
1642
1643 Ok(Self {
1644 parquet_column_index: Some(parquet_column_index),
1645 arrow_field,
1646 missing_null_counts_as_zero: true,
1647 physical_type: Some(physical_type),
1648 })
1649 }
1650
1651 pub fn row_group_mins<I>(&self, metadatas: I) -> Result<ArrayRef>
1695 where
1696 I: IntoIterator<Item = &'a RowGroupMetaData>,
1697 {
1698 let data_type = self.arrow_field.data_type();
1699
1700 let Some(parquet_index) = self.parquet_column_index else {
1701 return Ok(self.make_null_array(data_type, metadatas));
1702 };
1703
1704 let iter = metadatas
1705 .into_iter()
1706 .map(|x| x.column(parquet_index).statistics());
1707 min_statistics(data_type, iter, self.physical_type)
1708 }
1709
1710 pub fn row_group_maxes<I>(&self, metadatas: I) -> Result<ArrayRef>
1714 where
1715 I: IntoIterator<Item = &'a RowGroupMetaData>,
1716 {
1717 let data_type = self.arrow_field.data_type();
1718
1719 let Some(parquet_index) = self.parquet_column_index else {
1720 return Ok(self.make_null_array(data_type, metadatas));
1721 };
1722
1723 let iter = metadatas
1724 .into_iter()
1725 .map(|x| x.column(parquet_index).statistics());
1726 max_statistics(data_type, iter, self.physical_type)
1727 }
1728
1729 pub fn row_group_is_max_value_exact<I>(&self, metadatas: I) -> Result<BooleanArray>
1733 where
1734 I: IntoIterator<Item = &'a RowGroupMetaData>,
1735 {
1736 let Some(parquet_index) = self.parquet_column_index else {
1737 let num_row_groups = metadatas.into_iter().count();
1738 return Ok(BooleanArray::from_iter(std::iter::repeat_n(
1739 None,
1740 num_row_groups,
1741 )));
1742 };
1743
1744 let is_max_value_exact = metadatas
1745 .into_iter()
1746 .map(|x| x.column(parquet_index).statistics())
1747 .map(|s| s.map(|s| s.max_is_exact()));
1748 Ok(BooleanArray::from_iter(is_max_value_exact))
1749 }
1750
1751 pub fn row_group_is_min_value_exact<I>(&self, metadatas: I) -> Result<BooleanArray>
1755 where
1756 I: IntoIterator<Item = &'a RowGroupMetaData>,
1757 {
1758 let Some(parquet_index) = self.parquet_column_index else {
1759 let num_row_groups = metadatas.into_iter().count();
1760 return Ok(BooleanArray::from_iter(std::iter::repeat_n(
1761 None,
1762 num_row_groups,
1763 )));
1764 };
1765
1766 let is_min_value_exact = metadatas
1767 .into_iter()
1768 .map(|x| x.column(parquet_index).statistics())
1769 .map(|s| s.map(|s| s.min_is_exact()));
1770 Ok(BooleanArray::from_iter(is_min_value_exact))
1771 }
1772
1773 pub fn row_group_null_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
1777 where
1778 I: IntoIterator<Item = &'a RowGroupMetaData>,
1779 {
1780 let Some(parquet_index) = self.parquet_column_index else {
1781 let num_row_groups = metadatas.into_iter().count();
1782 return Ok(UInt64Array::from_iter(std::iter::repeat_n(
1783 None,
1784 num_row_groups,
1785 )));
1786 };
1787
1788 let null_counts = metadatas
1789 .into_iter()
1790 .map(|x| x.column(parquet_index).statistics())
1791 .map(|s| {
1792 let s = s?;
1793 if self.missing_null_counts_as_zero {
1794 Some(s.null_count_opt().unwrap_or(0))
1795 } else {
1796 s.null_count_opt()
1797 }
1798 });
1799 Ok(UInt64Array::from_iter(null_counts))
1800 }
1801
1802 pub fn row_group_nan_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
1806 where
1807 I: IntoIterator<Item = &'a RowGroupMetaData>,
1808 {
1809 let Some(parquet_index) = self.parquet_column_index else {
1810 let num_row_groups = metadatas.into_iter().count();
1811 return Ok(UInt64Array::from_iter(std::iter::repeat_n(
1812 None,
1813 num_row_groups,
1814 )));
1815 };
1816
1817 let nan_counts = metadatas
1818 .into_iter()
1819 .map(|x| x.column(parquet_index).statistics())
1820 .map(|s| s?.nan_count_opt());
1821 Ok(UInt64Array::from_iter(nan_counts))
1822 }
1823
1824 pub fn row_group_distinct_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
1828 where
1829 I: IntoIterator<Item = &'a RowGroupMetaData>,
1830 {
1831 let Some(parquet_index) = self.parquet_column_index else {
1832 let num_row_groups = metadatas.into_iter().count();
1833 return Ok(UInt64Array::from_iter(std::iter::repeat_n(
1834 None,
1835 num_row_groups,
1836 )));
1837 };
1838
1839 let distinct_counts = metadatas
1840 .into_iter()
1841 .map(|x| x.column(parquet_index).statistics())
1842 .map(|s| s?.distinct_count_opt());
1843 Ok(UInt64Array::from_iter(distinct_counts))
1844 }
1845
1846 pub fn data_page_mins<I>(
1896 &self,
1897 page_index: &dyn PageIndexProvider,
1898 row_group_indices: I,
1899 ) -> Result<ArrayRef>
1900 where
1901 I: IntoIterator<Item = &'a usize>,
1902 {
1903 let data_type = self.arrow_field.data_type();
1904
1905 let Some(parquet_index) = self.parquet_column_index else {
1906 return Ok(self.make_null_array(data_type, row_group_indices));
1907 };
1908
1909 let iter = row_group_indices.into_iter().map(|rg_index| {
1910 let column_page_index_per_row_group_per_column =
1911 page_index.column_index(*rg_index, parquet_index);
1912 let num_data_pages = page_index
1913 .num_data_pages(*rg_index, parquet_index)
1914 .unwrap_or(0);
1915
1916 (num_data_pages, column_page_index_per_row_group_per_column)
1917 });
1918
1919 min_page_statistics(data_type, iter, self.physical_type)
1920 }
1921
1922 pub fn data_page_maxes<I>(
1926 &self,
1927 page_index: &dyn PageIndexProvider,
1928 row_group_indices: I,
1929 ) -> Result<ArrayRef>
1930 where
1931 I: IntoIterator<Item = &'a usize>,
1932 {
1933 let data_type = self.arrow_field.data_type();
1934
1935 let Some(parquet_index) = self.parquet_column_index else {
1936 return Ok(self.make_null_array(data_type, row_group_indices));
1937 };
1938
1939 let iter = row_group_indices.into_iter().map(|rg_index| {
1940 let column_page_index_per_row_group_per_column =
1941 page_index.column_index(*rg_index, parquet_index);
1942 let num_data_pages = page_index
1943 .num_data_pages(*rg_index, parquet_index)
1944 .unwrap_or(0);
1945
1946 (num_data_pages, column_page_index_per_row_group_per_column)
1947 });
1948
1949 max_page_statistics(data_type, iter, self.physical_type)
1950 }
1951
1952 pub fn data_page_null_counts<I>(
1956 &self,
1957 page_index: &dyn PageIndexProvider,
1958 row_group_indices: I,
1959 ) -> Result<UInt64Array>
1960 where
1961 I: IntoIterator<Item = &'a usize>,
1962 {
1963 let Some(parquet_index) = self.parquet_column_index else {
1964 let num_row_groups = row_group_indices.into_iter().count();
1965 return Ok(UInt64Array::new_null(num_row_groups));
1966 };
1967
1968 let iter = row_group_indices.into_iter().map(|rg_index| {
1969 let column_page_index_per_row_group_per_column =
1970 page_index.column_index(*rg_index, parquet_index);
1971 let num_data_pages = page_index
1972 .num_data_pages(*rg_index, parquet_index)
1973 .unwrap_or(0);
1974
1975 (num_data_pages, column_page_index_per_row_group_per_column)
1976 });
1977 null_counts_page_statistics(iter)
1978 }
1979
1980 pub fn data_page_nan_counts<I>(
1984 &self,
1985 page_index: &dyn PageIndexProvider,
1986 row_group_indices: I,
1987 ) -> Result<UInt64Array>
1988 where
1989 I: IntoIterator<Item = &'a usize>,
1990 {
1991 let Some(parquet_index) = self.parquet_column_index else {
1992 let num_row_groups = row_group_indices.into_iter().count();
1993 return Ok(UInt64Array::new_null(num_row_groups));
1994 };
1995
1996 let iter = row_group_indices.into_iter().map(|rg_index| {
1997 let column_page_index_per_row_group_per_column =
1998 page_index.column_index(*rg_index, parquet_index);
1999 let num_data_pages = page_index
2000 .num_data_pages(*rg_index, parquet_index)
2001 .unwrap_or(0);
2002
2003 (num_data_pages, column_page_index_per_row_group_per_column)
2004 });
2005 nan_counts_page_statistics(iter)
2006 }
2007
2008 pub fn data_page_row_counts<I>(
2026 &self,
2027 page_index: &dyn PageIndexProvider,
2028 row_group_metadatas: &'a [RowGroupMetaData],
2029 row_group_indices: I,
2030 ) -> Result<Option<UInt64Array>>
2031 where
2032 I: IntoIterator<Item = &'a usize>,
2033 {
2034 let Some(parquet_index) = self.parquet_column_index else {
2035 return Ok(None);
2039 };
2040
2041 let mut row_counts = Vec::new();
2042 let mut nulls = NullBufferBuilder::new(0);
2043 for rg_idx in row_group_indices {
2044 let Some(offset_index) = page_index.offset_index(*rg_idx, parquet_index) else {
2045 continue;
2046 };
2047 let page_locations = offset_index.page_locations();
2048
2049 let row_count_per_page = page_locations
2050 .windows(2)
2051 .map(|loc| Some(loc[1].first_row_index as u64 - loc[0].first_row_index as u64));
2052
2053 let num_rows_in_row_group = &row_group_metadatas[*rg_idx].num_rows();
2055 let row_count_per_page = row_count_per_page.chain(std::iter::once(Some(
2056 *num_rows_in_row_group as u64
2057 - page_locations.last().unwrap().first_row_index as u64,
2058 )));
2059
2060 row_counts.extend(row_count_per_page.clone().map(|x| x.unwrap_or(0)));
2061 for val in row_count_per_page {
2062 if val.is_some() {
2063 nulls.append_non_null();
2064 } else {
2065 nulls.append_null();
2066 }
2067 }
2068 }
2069
2070 Ok(Some(UInt64Array::new(row_counts.into(), nulls.build())))
2071 }
2072
2073 fn make_null_array<I, A>(&self, data_type: &DataType, metadatas: I) -> ArrayRef
2075 where
2076 I: IntoIterator<Item = A>,
2077 {
2078 let num_row_groups = metadatas.into_iter().count();
2080 new_null_array(data_type, num_row_groups)
2081 }
2082}
2083
2084