1use crate::basic::{PageType, Type};
22use crate::bloom_filter::Sbbf;
23use crate::column::page::{Page, PageMetadata, PageReader};
24use crate::compression::{Codec, create_codec};
25#[cfg(feature = "encryption")]
26use crate::encryption::decrypt::{CryptoContext, read_and_decrypt};
27use crate::errors::{ParquetError, Result};
28use crate::file::metadata::page_index::RowGroupPageIndex;
29use crate::file::metadata::thrift::PageHeader;
30use crate::file::page_index::offset_index::PageLocation;
31use crate::file::statistics;
32use crate::file::{
33 metadata::*,
34 properties::{ReaderProperties, ReaderPropertiesPtr},
35 reader::*,
36};
37#[cfg(feature = "encryption")]
38use crate::parquet_thrift::ThriftSliceInputProtocol;
39use crate::parquet_thrift::{ReadThrift, ThriftReadInputProtocol};
40use crate::record::Row;
41use crate::record::reader::RowIter;
42use crate::schema::types::{SchemaDescPtr, Type as SchemaType};
43use bytes::Bytes;
44use std::collections::VecDeque;
45use std::{fs::File, io::Read, path::Path, sync::Arc};
46
47impl TryFrom<File> for SerializedFileReader<File> {
48 type Error = ParquetError;
49
50 fn try_from(file: File) -> Result<Self> {
51 Self::new(file)
52 }
53}
54
55impl TryFrom<&Path> for SerializedFileReader<File> {
56 type Error = ParquetError;
57
58 fn try_from(path: &Path) -> Result<Self> {
59 let file = File::open(path)?;
60 Self::try_from(file)
61 }
62}
63
64impl TryFrom<String> for SerializedFileReader<File> {
65 type Error = ParquetError;
66
67 fn try_from(path: String) -> Result<Self> {
68 Self::try_from(Path::new(&path))
69 }
70}
71
72impl TryFrom<&str> for SerializedFileReader<File> {
73 type Error = ParquetError;
74
75 fn try_from(path: &str) -> Result<Self> {
76 Self::try_from(Path::new(&path))
77 }
78}
79
80impl IntoIterator for SerializedFileReader<File> {
83 type Item = Result<Row>;
84 type IntoIter = RowIter<'static>;
85
86 fn into_iter(self) -> Self::IntoIter {
87 RowIter::from_file_into(Box::new(self))
88 }
89}
90
91pub struct SerializedFileReader<R: ChunkReader> {
96 chunk_reader: Arc<R>,
97 metadata: Arc<ParquetMetaData>,
98 props: ReaderPropertiesPtr,
99}
100
101pub type ReadGroupPredicate = Box<dyn FnMut(&RowGroupMetaData, usize) -> bool>;
105
106#[derive(Default)]
110pub struct ReadOptionsBuilder {
111 predicates: Vec<ReadGroupPredicate>,
112 enable_page_index: bool,
113 props: Option<ReaderProperties>,
114 metadata_options: ParquetMetaDataOptions,
115}
116
117impl ReadOptionsBuilder {
118 pub fn new() -> Self {
120 Self::default()
121 }
122
123 pub fn with_predicate(mut self, predicate: ReadGroupPredicate) -> Self {
126 self.predicates.push(predicate);
127 self
128 }
129
130 pub fn with_range(mut self, start: i64, end: i64) -> Self {
137 assert!(start < end);
138 let predicate = move |rg: &RowGroupMetaData, _: usize| {
139 let mid = get_midpoint_offset(rg);
140 mid >= start && mid < end
141 };
142 self.predicates.push(Box::new(predicate));
143 self
144 }
145
146 pub fn with_page_index(mut self) -> Self {
151 self.enable_page_index = true;
152 self
153 }
154
155 pub fn with_reader_properties(mut self, properties: ReaderProperties) -> Self {
157 self.props = Some(properties);
158 self
159 }
160
161 pub fn with_parquet_schema(mut self, schema: SchemaDescPtr) -> Self {
164 self.metadata_options.set_schema(schema);
165 self
166 }
167
168 pub fn with_encoding_stats_as_mask(mut self, val: bool) -> Self {
177 self.metadata_options.set_encoding_stats_as_mask(val);
178 self
179 }
180
181 pub fn with_encoding_stats_policy(mut self, policy: ParquetStatisticsPolicy) -> Self {
186 self.metadata_options.set_encoding_stats_policy(policy);
187 self
188 }
189
190 pub fn with_column_stats_policy(mut self, policy: ParquetStatisticsPolicy) -> Self {
195 self.metadata_options.set_column_stats_policy(policy);
196 self
197 }
198
199 pub fn with_size_stats_policy(mut self, policy: ParquetStatisticsPolicy) -> Self {
204 self.metadata_options.set_size_stats_policy(policy);
205 self
206 }
207
208 pub fn build(self) -> ReadOptions {
210 let props = self
211 .props
212 .unwrap_or_else(|| ReaderProperties::builder().build());
213 ReadOptions {
214 predicates: self.predicates,
215 enable_page_index: self.enable_page_index,
216 props,
217 metadata_options: self.metadata_options,
218 }
219 }
220}
221
222pub struct ReadOptions {
227 predicates: Vec<ReadGroupPredicate>,
228 enable_page_index: bool,
229 props: ReaderProperties,
230 metadata_options: ParquetMetaDataOptions,
231}
232
233impl<R: 'static + ChunkReader> SerializedFileReader<R> {
234 pub fn new(chunk_reader: R) -> Result<Self> {
237 let metadata = ParquetMetaDataReader::new().parse_and_finish(&chunk_reader)?;
238 let props = Arc::new(ReaderProperties::builder().build());
239 Ok(Self {
240 chunk_reader: Arc::new(chunk_reader),
241 metadata: Arc::new(metadata),
242 props,
243 })
244 }
245
246 pub fn new_with_options(chunk_reader: R, options: ReadOptions) -> Result<Self> {
249 let mut metadata_builder = ParquetMetaDataReader::new()
250 .with_metadata_options(Some(options.metadata_options.clone()))
251 .parse_and_finish(&chunk_reader)?
252 .into_builder();
253 let mut predicates = options.predicates;
254
255 for (i, rg_meta) in metadata_builder.take_row_groups().into_iter().enumerate() {
257 let mut keep = true;
258 for predicate in &mut predicates {
259 if !predicate(&rg_meta, i) {
260 keep = false;
261 break;
262 }
263 }
264 if keep {
265 metadata_builder = metadata_builder.add_row_group(rg_meta);
266 }
267 }
268
269 let mut metadata = metadata_builder.build();
270
271 if options.enable_page_index {
273 let mut reader = ParquetMetaDataReader::new_with_metadata(metadata)
274 .with_page_index_policy(PageIndexPolicy::Required);
275 reader.read_page_indexes(&chunk_reader)?;
276 metadata = reader.finish()?;
277 }
278
279 Ok(Self {
280 chunk_reader: Arc::new(chunk_reader),
281 metadata: Arc::new(metadata),
282 props: Arc::new(options.props),
283 })
284 }
285}
286
287fn get_midpoint_offset(meta: &RowGroupMetaData) -> i64 {
289 let col = meta.column(0);
290 let mut offset = col.data_page_offset();
291 if let Some(dic_offset) = col.dictionary_page_offset()
292 && offset > dic_offset
293 {
294 offset = dic_offset
295 }
296 offset + meta.compressed_size() / 2
297}
298
299impl<R: 'static + ChunkReader> FileReader for SerializedFileReader<R> {
300 fn metadata(&self) -> &ParquetMetaData {
301 &self.metadata
302 }
303
304 fn num_row_groups(&self) -> usize {
305 self.metadata.num_row_groups()
306 }
307
308 fn get_row_group(&self, i: usize) -> Result<Box<dyn RowGroupReader + '_>> {
309 let row_group_metadata = self.metadata.row_group(i);
310 let props = Arc::clone(&self.props);
312 let f = Arc::clone(&self.chunk_reader);
313 let page_index = self.metadata.page_index_for_row_group(i);
314 Ok(Box::new(SerializedRowGroupReader::new(
315 f,
316 row_group_metadata,
317 page_index,
318 props,
319 )?))
320 }
321
322 fn get_row_iter(&self, projection: Option<SchemaType>) -> Result<RowIter<'_>> {
323 RowIter::from_file(projection, self)
324 }
325}
326
327pub struct SerializedRowGroupReader<'a, R: ChunkReader> {
329 chunk_reader: Arc<R>,
330 metadata: &'a RowGroupMetaData,
331 page_index: RowGroupPageIndex,
332 props: ReaderPropertiesPtr,
333 bloom_filters: Vec<Option<Sbbf>>,
334}
335
336impl<'a, R: ChunkReader> SerializedRowGroupReader<'a, R> {
337 pub fn new(
339 chunk_reader: Arc<R>,
340 metadata: &'a RowGroupMetaData,
341 page_index: RowGroupPageIndex,
342 props: ReaderPropertiesPtr,
343 ) -> Result<Self> {
344 let bloom_filters = if props.read_bloom_filter() {
345 metadata
346 .columns()
347 .iter()
348 .map(|col| Sbbf::read_from_column_chunk(col, &*chunk_reader))
349 .collect::<Result<Vec<_>>>()?
350 } else {
351 std::iter::repeat_n(None, metadata.columns().len()).collect()
352 };
353 Ok(Self {
354 chunk_reader,
355 metadata,
356 page_index,
357 props,
358 bloom_filters,
359 })
360 }
361}
362
363impl<R: 'static + ChunkReader> RowGroupReader for SerializedRowGroupReader<'_, R> {
364 fn metadata(&self) -> &RowGroupMetaData {
365 self.metadata
366 }
367
368 fn num_columns(&self) -> usize {
369 self.metadata.num_columns()
370 }
371
372 fn get_column_page_reader(&self, i: usize) -> Result<Box<dyn PageReader>> {
374 let col = self.metadata.column(i);
375
376 let page_locations = self.page_index.page_locations(i).cloned();
378
379 let props = Arc::clone(&self.props);
380 Ok(Box::new(SerializedPageReader::new_with_properties(
381 Arc::clone(&self.chunk_reader),
382 col,
383 usize::try_from(self.metadata.num_rows())?,
384 page_locations,
385 props,
386 )?))
387 }
388
389 fn get_column_bloom_filter(&self, i: usize) -> Option<&Sbbf> {
391 self.bloom_filters[i].as_ref()
392 }
393
394 fn get_row_iter(&self, projection: Option<SchemaType>) -> Result<RowIter<'_>> {
395 RowIter::from_row_group(projection, self)
396 }
397}
398
399pub(crate) fn decode_page(
401 page_header: PageHeader,
402 buffer: Bytes,
403 physical_type: Type,
404 decompressor: Option<&mut Box<dyn Codec>>,
405) -> Result<Page> {
406 #[cfg(feature = "crc")]
408 if let Some(expected_crc) = page_header.crc {
409 let crc = crc32fast::hash(&buffer);
410 if crc != expected_crc as u32 {
411 return Err(general_err!("Page CRC checksum mismatch"));
412 }
413 }
414
415 let (offset, can_decompress): (usize, bool) = match page_header.data_page_header_v2 {
422 Some(ref header_v2) => {
423 if header_v2.definition_levels_byte_length < 0
424 || header_v2.repetition_levels_byte_length < 0
425 || header_v2.definition_levels_byte_length + header_v2.repetition_levels_byte_length
426 > page_header.uncompressed_page_size
427 {
428 return Err(general_err!(
429 "DataPage v2 header contains implausible values \
430 for definition_levels_byte_length ({}) \
431 and repetition_levels_byte_length ({}) \
432 given DataPage header provides uncompressed_page_size ({})",
433 header_v2.definition_levels_byte_length,
434 header_v2.repetition_levels_byte_length,
435 page_header.uncompressed_page_size
436 ));
437 }
438 (
439 usize::try_from(
440 header_v2.definition_levels_byte_length
441 + header_v2.repetition_levels_byte_length,
442 )?,
443 header_v2.is_compressed.unwrap_or(true),
445 )
446 }
447 None => (0, true),
448 };
449
450 let buffer = match decompressor {
451 Some(decompressor) if can_decompress => {
452 let uncompressed_page_size = usize::try_from(page_header.uncompressed_page_size)?;
453 if offset > buffer.len() || offset > uncompressed_page_size {
454 return Err(general_err!("Invalid page header"));
455 }
456 let decompressed_size = uncompressed_page_size - offset;
457 let mut decompressed = Vec::with_capacity(uncompressed_page_size);
458 decompressed.extend_from_slice(&buffer[..offset]);
459 if decompressed_size > 0 {
462 let compressed = &buffer[offset..];
463 decompressor.decompress(compressed, &mut decompressed, Some(decompressed_size))?;
464 }
465
466 if decompressed.len() != uncompressed_page_size {
467 return Err(general_err!(
468 "Actual decompressed size doesn't match the expected one ({} vs {})",
469 decompressed.len(),
470 uncompressed_page_size
471 ));
472 }
473
474 Bytes::from(decompressed)
475 }
476 _ => buffer,
477 };
478
479 let result = match page_header.r#type {
480 PageType::DICTIONARY_PAGE => {
481 let dict_header = page_header.dictionary_page_header.as_ref().ok_or_else(|| {
482 ParquetError::General("Missing dictionary page header".to_string())
483 })?;
484 let is_sorted = dict_header.is_sorted.unwrap_or(false);
485 Page::DictionaryPage {
486 buf: buffer,
487 num_values: dict_header.num_values.try_into()?,
488 encoding: dict_header.encoding,
489 is_sorted,
490 }
491 }
492 PageType::DATA_PAGE => {
493 let header = page_header
494 .data_page_header
495 .ok_or_else(|| ParquetError::General("Missing V1 data page header".to_string()))?;
496 Page::DataPage {
497 buf: buffer,
498 num_values: header.num_values.try_into()?,
499 encoding: header.encoding,
500 def_level_encoding: header.definition_level_encoding,
501 rep_level_encoding: header.repetition_level_encoding,
502 statistics: statistics::from_thrift_page_stats(physical_type, header.statistics)?,
503 }
504 }
505 PageType::DATA_PAGE_V2 => {
506 let header = page_header
507 .data_page_header_v2
508 .ok_or_else(|| ParquetError::General("Missing V2 data page header".to_string()))?;
509 let is_compressed = header.is_compressed.unwrap_or(true);
510 Page::DataPageV2 {
511 buf: buffer,
512 num_values: header.num_values.try_into()?,
513 encoding: header.encoding,
514 num_nulls: header.num_nulls.try_into()?,
515 num_rows: header.num_rows.try_into()?,
516 def_levels_byte_len: header.definition_levels_byte_length.try_into()?,
517 rep_levels_byte_len: header.repetition_levels_byte_length.try_into()?,
518 is_compressed,
519 statistics: statistics::from_thrift_page_stats(physical_type, header.statistics)?,
520 }
521 }
522 PageType::INDEX_PAGE => {
523 return Err(general_err!(
525 "Page type {:?} is not supported",
526 page_header.r#type
527 ));
528 }
529 };
530
531 Ok(result)
532}
533
534enum SerializedPageReaderState {
535 Values {
536 offset: u64,
539
540 remaining_bytes: u64,
543
544 next_page_header: Option<Box<PageHeader>>,
546
547 page_index: usize,
549
550 require_dictionary: bool,
552 },
553 Pages {
554 page_locations: VecDeque<PageLocation>,
556 dictionary_page: Option<PageLocation>,
558 total_rows: usize,
560 page_index: usize,
562 },
563}
564
565#[derive(Default)]
566pub(crate) struct SerializedPageReaderContext {
567 pub(crate) read_stats: bool,
569 #[cfg(feature = "encryption")]
571 pub(crate) crypto_context: Option<Arc<CryptoContext>>,
572}
573
574pub struct SerializedPageReader<R: ChunkReader> {
576 reader: Arc<R>,
578
579 decompressor: Option<Box<dyn Codec>>,
581
582 physical_type: Type,
584
585 state: SerializedPageReaderState,
586
587 context: SerializedPageReaderContext,
588}
589
590impl<R: ChunkReader> SerializedPageReader<R> {
591 pub fn new(
593 reader: Arc<R>,
594 column_chunk_metadata: &ColumnChunkMetaData,
595 total_rows: usize,
596 page_locations: Option<Vec<PageLocation>>,
597 ) -> Result<Self> {
598 let props = Arc::new(ReaderProperties::builder().build());
599 SerializedPageReader::new_with_properties(
600 reader,
601 column_chunk_metadata,
602 total_rows,
603 page_locations,
604 props,
605 )
606 }
607
608 #[cfg(all(feature = "arrow", not(feature = "encryption")))]
610 pub(crate) fn add_crypto_context(
611 self,
612 _rg_idx: usize,
613 _column_idx: usize,
614 _parquet_meta_data: &ParquetMetaData,
615 _column_chunk_metadata: &ColumnChunkMetaData,
616 ) -> Result<SerializedPageReader<R>> {
617 Ok(self)
618 }
619
620 #[cfg(feature = "encryption")]
622 pub(crate) fn add_crypto_context(
623 mut self,
624 rg_idx: usize,
625 column_idx: usize,
626 parquet_meta_data: &ParquetMetaData,
627 column_chunk_metadata: &ColumnChunkMetaData,
628 ) -> Result<SerializedPageReader<R>> {
629 let Some(file_decryptor) = parquet_meta_data.file_decryptor() else {
630 return Ok(self);
631 };
632 let Some(crypto_metadata) = column_chunk_metadata.crypto_metadata() else {
633 return Ok(self);
634 };
635 let crypto_context =
636 CryptoContext::for_column(file_decryptor, crypto_metadata, rg_idx, column_idx)?;
637 self.context.crypto_context = Some(Arc::new(crypto_context));
638 Ok(self)
639 }
640
641 pub fn new_with_properties(
643 reader: Arc<R>,
644 meta: &ColumnChunkMetaData,
645 total_rows: usize,
646 page_locations: Option<Vec<PageLocation>>,
647 props: ReaderPropertiesPtr,
648 ) -> Result<Self> {
649 let decompressor = create_codec(meta.compression(), props.codec_options())?;
650 let (start, len) = meta.byte_range();
651
652 let state = match page_locations {
653 Some(locations) => {
654 let dictionary_page = match locations.first() {
657 Some(dict_offset) if dict_offset.offset as u64 != start => Some(PageLocation {
658 offset: start as i64,
659 compressed_page_size: (dict_offset.offset as u64 - start) as i32,
660 first_row_index: 0,
661 }),
662 _ => None,
663 };
664
665 SerializedPageReaderState::Pages {
666 page_locations: locations.into(),
667 dictionary_page,
668 total_rows,
669 page_index: 0,
670 }
671 }
672 None => SerializedPageReaderState::Values {
673 offset: start,
674 remaining_bytes: len,
675 next_page_header: None,
676 page_index: 0,
677 require_dictionary: meta.dictionary_page_offset().is_some(),
678 },
679 };
680 let mut context = SerializedPageReaderContext::default();
681 if props.read_page_stats() {
682 context.read_stats = true;
683 }
684 Ok(Self {
685 reader,
686 decompressor,
687 state,
688 physical_type: meta.column_type(),
689 context,
690 })
691 }
692
693 #[cfg(test)]
699 fn peek_next_page_offset(&mut self) -> Result<Option<u64>> {
700 match &mut self.state {
701 SerializedPageReaderState::Values {
702 offset,
703 remaining_bytes,
704 next_page_header,
705 page_index,
706 require_dictionary,
707 } => {
708 loop {
709 if *remaining_bytes == 0 {
710 return Ok(None);
711 }
712 return if let Some(header) = next_page_header.as_ref() {
713 if let Ok(_page_meta) = PageMetadata::try_from(&**header) {
714 Ok(Some(*offset))
715 } else {
716 *next_page_header = None;
718 continue;
719 }
720 } else {
721 let mut read = self.reader.get_read(*offset)?;
722 let (header_len, header) = Self::read_page_header_len(
723 &self.context,
724 &mut read,
725 *page_index,
726 *require_dictionary,
727 )?;
728 *offset += header_len as u64;
729 *remaining_bytes -= header_len as u64;
730 let page_meta = if let Ok(_page_meta) = PageMetadata::try_from(&header) {
731 Ok(Some(*offset))
732 } else {
733 continue;
735 };
736 *next_page_header = Some(Box::new(header));
737 page_meta
738 };
739 }
740 }
741 SerializedPageReaderState::Pages {
742 page_locations,
743 dictionary_page,
744 ..
745 } => {
746 if let Some(page) = dictionary_page {
747 Ok(Some(page.offset as u64))
748 } else if let Some(page) = page_locations.front() {
749 Ok(Some(page.offset as u64))
750 } else {
751 Ok(None)
752 }
753 }
754 }
755 }
756
757 fn read_page_header_len<T: Read>(
758 context: &SerializedPageReaderContext,
759 input: &mut T,
760 page_index: usize,
761 dictionary_page: bool,
762 ) -> Result<(usize, PageHeader)> {
763 struct TrackedRead<R> {
765 inner: R,
766 bytes_read: usize,
767 }
768
769 impl<R: Read> Read for TrackedRead<R> {
770 fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
771 let v = self.inner.read(buf)?;
772 self.bytes_read += v;
773 Ok(v)
774 }
775 }
776
777 let mut tracked = TrackedRead {
778 inner: input,
779 bytes_read: 0,
780 };
781 let header = context.read_page_header(&mut tracked, page_index, dictionary_page)?;
782 Ok((tracked.bytes_read, header))
783 }
784}
785
786pub(crate) fn read_page_header_len_from_bytes(
794 context: &SerializedPageReaderContext,
795 buffer: &[u8],
796 page_index: usize,
797 dictionary_page: bool,
798) -> Result<(usize, PageHeader)> {
799 let mut input = std::io::Cursor::new(buffer);
800 let header = context.read_page_header(&mut input, page_index, dictionary_page)?;
801 let header_len = input.position() as usize;
802 Ok((header_len, header))
803}
804
805#[cfg(not(feature = "encryption"))]
806impl SerializedPageReaderContext {
807 pub(crate) fn read_page_header<T: Read>(
808 &self,
809 input: &mut T,
810 _page_index: usize,
811 _dictionary_page: bool,
812 ) -> Result<PageHeader> {
813 let mut prot = ThriftReadInputProtocol::new(input);
814 if self.read_stats {
815 Ok(PageHeader::read_thrift(&mut prot)?)
816 } else {
817 Ok(PageHeader::read_thrift_without_stats(&mut prot)?)
818 }
819 }
820
821 pub(crate) fn decrypt_page_data<T>(
822 &self,
823 buffer: T,
824 _page_index: usize,
825 _dictionary_page: bool,
826 ) -> Result<T> {
827 Ok(buffer)
828 }
829}
830
831#[cfg(feature = "encryption")]
832impl SerializedPageReaderContext {
833 pub(crate) fn read_page_header<T: Read>(
834 &self,
835 input: &mut T,
836 page_index: usize,
837 dictionary_page: bool,
838 ) -> Result<PageHeader> {
839 match self.page_crypto_context(page_index, dictionary_page) {
840 None => {
841 let mut prot = ThriftReadInputProtocol::new(input);
842 if self.read_stats {
843 Ok(PageHeader::read_thrift(&mut prot)?)
844 } else {
845 use crate::file::metadata::thrift::PageHeader;
846
847 Ok(PageHeader::read_thrift_without_stats(&mut prot)?)
848 }
849 }
850 Some(page_crypto_context) => {
851 let data_decryptor = page_crypto_context.data_decryptor();
852 let aad = page_crypto_context.create_page_header_aad()?;
853
854 let buf = read_and_decrypt(data_decryptor, input, aad.as_ref()).map_err(|_| {
855 ParquetError::General(format!(
856 "Error decrypting page header for column {}, decryption key may be wrong",
857 page_crypto_context.column_ordinal
858 ))
859 })?;
860
861 let mut prot = ThriftSliceInputProtocol::new(buf.as_slice());
862 if self.read_stats {
863 Ok(PageHeader::read_thrift(&mut prot)?)
864 } else {
865 Ok(PageHeader::read_thrift_without_stats(&mut prot)?)
866 }
867 }
868 }
869 }
870
871 pub(crate) fn decrypt_page_data<T>(
872 &self,
873 buffer: T,
874 page_index: usize,
875 dictionary_page: bool,
876 ) -> Result<T>
877 where
878 T: AsRef<[u8]>,
879 T: From<Vec<u8>>,
880 {
881 let page_crypto_context = self.page_crypto_context(page_index, dictionary_page);
882 if let Some(page_crypto_context) = page_crypto_context {
883 let decryptor = page_crypto_context.data_decryptor();
884 let aad = page_crypto_context.create_page_aad()?;
885 let decrypted = decryptor.decrypt(buffer.as_ref(), &aad)?;
886 Ok(T::from(decrypted))
887 } else {
888 Ok(buffer)
889 }
890 }
891
892 fn page_crypto_context(
893 &self,
894 page_index: usize,
895 dictionary_page: bool,
896 ) -> Option<Arc<CryptoContext>> {
897 self.crypto_context.as_ref().map(|c| {
898 Arc::new(if dictionary_page {
899 c.for_dictionary_page()
900 } else {
901 c.with_page_ordinal(page_index)
902 })
903 })
904 }
905}
906
907impl<R: ChunkReader> Iterator for SerializedPageReader<R> {
908 type Item = Result<Page>;
909
910 fn next(&mut self) -> Option<Self::Item> {
911 self.get_next_page().transpose()
912 }
913}
914
915fn verify_page_header_len(header_len: usize, remaining_bytes: u64) -> Result<()> {
916 if header_len as u64 > remaining_bytes {
917 return Err(eof_err!("Invalid page header"));
918 }
919 Ok(())
920}
921
922pub(crate) fn verify_page_size(
923 compressed_size: i32,
924 uncompressed_size: i32,
925 remaining_bytes: u64,
926) -> Result<()> {
927 if compressed_size < 0 || compressed_size as u64 > remaining_bytes || uncompressed_size < 0 {
931 return Err(eof_err!("Invalid page header"));
932 }
933 Ok(())
934}
935
936impl<R: ChunkReader> PageReader for SerializedPageReader<R> {
937 fn get_next_page(&mut self) -> Result<Option<Page>> {
938 loop {
939 let page = match &mut self.state {
940 SerializedPageReaderState::Values {
941 offset,
942 remaining_bytes: remaining,
943 next_page_header,
944 page_index,
945 require_dictionary,
946 } => {
947 if *remaining == 0 {
948 return Ok(None);
949 }
950
951 let mut read = self.reader.get_read(*offset)?;
952 let header = if let Some(header) = next_page_header.take() {
953 *header
954 } else {
955 let (header_len, header) = Self::read_page_header_len(
956 &self.context,
957 &mut read,
958 *page_index,
959 *require_dictionary,
960 )?;
961 verify_page_header_len(header_len, *remaining)?;
962 *offset += header_len as u64;
963 *remaining -= header_len as u64;
964 header
965 };
966 verify_page_size(
967 header.compressed_page_size,
968 header.uncompressed_page_size,
969 *remaining,
970 )?;
971 let data_len = header.compressed_page_size as usize;
972 let data_start = *offset;
973 *offset += data_len as u64;
974 *remaining -= data_len as u64;
975
976 if header.r#type == PageType::INDEX_PAGE {
977 continue;
978 }
979
980 let buffer = self.reader.get_bytes(data_start, data_len)?;
981
982 let buffer =
983 self.context
984 .decrypt_page_data(buffer, *page_index, *require_dictionary)?;
985
986 let page = decode_page(
987 header,
988 buffer,
989 self.physical_type,
990 self.decompressor.as_mut(),
991 )?;
992 if page.is_data_page() {
993 *page_index += 1;
994 } else if page.is_dictionary_page() {
995 *require_dictionary = false;
996 }
997 page
998 }
999 SerializedPageReaderState::Pages {
1000 page_locations,
1001 dictionary_page,
1002 page_index,
1003 ..
1004 } => {
1005 let (front, is_dictionary_page) = match dictionary_page.take() {
1006 Some(front) => (front, true),
1007 None => match page_locations.pop_front() {
1008 Some(front) => (front, false),
1009 None => return Ok(None),
1010 },
1011 };
1012
1013 let page_len = usize::try_from(front.compressed_page_size)?;
1014 let buffer = self.reader.get_bytes(front.offset as u64, page_len)?;
1015
1016 let (offset, header) = read_page_header_len_from_bytes(
1017 &self.context,
1018 buffer.as_ref(),
1019 *page_index,
1020 is_dictionary_page,
1021 )?;
1022 let bytes = buffer.slice(offset..);
1023 let bytes =
1024 self.context
1025 .decrypt_page_data(bytes, *page_index, is_dictionary_page)?;
1026
1027 if !is_dictionary_page {
1028 *page_index += 1;
1029 }
1030 decode_page(
1031 header,
1032 bytes,
1033 self.physical_type,
1034 self.decompressor.as_mut(),
1035 )?
1036 }
1037 };
1038
1039 return Ok(Some(page));
1040 }
1041 }
1042
1043 fn peek_next_page(&mut self) -> Result<Option<PageMetadata>> {
1044 match &mut self.state {
1045 SerializedPageReaderState::Values {
1046 offset,
1047 remaining_bytes,
1048 next_page_header,
1049 page_index,
1050 require_dictionary,
1051 } => {
1052 loop {
1053 if *remaining_bytes == 0 {
1054 return Ok(None);
1055 }
1056 return if let Some(header) = next_page_header.as_ref() {
1057 if let Ok(page_meta) = (&**header).try_into() {
1058 Ok(Some(page_meta))
1059 } else {
1060 *next_page_header = None;
1062 continue;
1063 }
1064 } else {
1065 let mut read = self.reader.get_read(*offset)?;
1066 let (header_len, header) = Self::read_page_header_len(
1067 &self.context,
1068 &mut read,
1069 *page_index,
1070 *require_dictionary,
1071 )?;
1072 verify_page_header_len(header_len, *remaining_bytes)?;
1073 *offset += header_len as u64;
1074 *remaining_bytes -= header_len as u64;
1075 let page_meta = if let Ok(page_meta) = (&header).try_into() {
1076 Ok(Some(page_meta))
1077 } else {
1078 continue;
1080 };
1081 *next_page_header = Some(Box::new(header));
1082 page_meta
1083 };
1084 }
1085 }
1086 SerializedPageReaderState::Pages {
1087 page_locations,
1088 dictionary_page,
1089 total_rows,
1090 page_index: _,
1091 } => {
1092 if dictionary_page.is_some() {
1093 Ok(Some(PageMetadata {
1094 num_rows: None,
1095 num_levels: None,
1096 is_dict: true,
1097 }))
1098 } else if let Some(page) = page_locations.front() {
1099 let next_rows = page_locations
1100 .get(1)
1101 .map(|x| x.first_row_index as usize)
1102 .unwrap_or(*total_rows);
1103
1104 Ok(Some(PageMetadata {
1105 num_rows: Some(next_rows - page.first_row_index as usize),
1106 num_levels: None,
1107 is_dict: false,
1108 }))
1109 } else {
1110 Ok(None)
1111 }
1112 }
1113 }
1114 }
1115
1116 fn skip_next_page(&mut self) -> Result<()> {
1117 match &mut self.state {
1118 SerializedPageReaderState::Values {
1119 offset,
1120 remaining_bytes,
1121 next_page_header,
1122 page_index,
1123 require_dictionary,
1124 } => {
1125 if let Some(buffered_header) = next_page_header.take() {
1126 verify_page_size(
1127 buffered_header.compressed_page_size,
1128 buffered_header.uncompressed_page_size,
1129 *remaining_bytes,
1130 )?;
1131 *offset += buffered_header.compressed_page_size as u64;
1133 *remaining_bytes -= buffered_header.compressed_page_size as u64;
1134 } else {
1135 let mut read = self.reader.get_read(*offset)?;
1136 let (header_len, header) = Self::read_page_header_len(
1137 &self.context,
1138 &mut read,
1139 *page_index,
1140 *require_dictionary,
1141 )?;
1142 verify_page_header_len(header_len, *remaining_bytes)?;
1143 verify_page_size(
1144 header.compressed_page_size,
1145 header.uncompressed_page_size,
1146 *remaining_bytes,
1147 )?;
1148 let data_page_size = header.compressed_page_size as u64;
1149 *offset += header_len as u64 + data_page_size;
1150 *remaining_bytes -= header_len as u64 + data_page_size;
1151 }
1152 if *require_dictionary {
1153 *require_dictionary = false;
1154 } else {
1155 *page_index += 1;
1156 }
1157 }
1158 SerializedPageReaderState::Pages {
1159 page_locations,
1160 dictionary_page,
1161 page_index,
1162 ..
1163 } => {
1164 if dictionary_page.is_some() {
1165 dictionary_page.take();
1167 } else {
1168 if page_locations.pop_front().is_some() {
1170 *page_index += 1;
1171 }
1172 }
1173 }
1174 }
1175 Ok(())
1176 }
1177
1178 fn at_record_boundary(&mut self) -> Result<bool> {
1179 match &mut self.state {
1180 SerializedPageReaderState::Values { .. } => match self.peek_next_page()? {
1181 None => Ok(true),
1182 Some(metadata) => Ok(metadata.num_rows.is_some()),
1185 },
1186 SerializedPageReaderState::Pages { .. } => Ok(true),
1187 }
1188 }
1189}
1190
1191#[cfg(test)]
1192mod tests {
1193 use std::collections::HashSet;
1194
1195 use bytes::Buf;
1196
1197 use crate::file::page_index::column_index::{
1198 ByteArrayColumnIndex, ColumnIndexMetaData, PrimitiveColumnIndex,
1199 };
1200 use crate::file::properties::{EnabledStatistics, WriterProperties};
1201
1202 use crate::basic::{self, BoundaryOrder, ColumnOrder, Encoding, SortOrder};
1203 use crate::column::reader::ColumnReader;
1204 use crate::data_type::private::ParquetValueType;
1205 use crate::data_type::{AsBytes, FixedLenByteArrayType, Int32Type};
1206 use crate::file::metadata::thrift::DataPageHeaderV2;
1207 use crate::file::writer::SerializedFileWriter;
1208 use crate::record::RowAccessor;
1209 use crate::schema::parser::parse_message_type;
1210 use crate::util::test_common::file_util::{get_test_file, get_test_path};
1211
1212 use super::*;
1213
1214 #[test]
1215 fn test_decode_page_invalid_offset() {
1216 let page_header = PageHeader {
1217 r#type: PageType::DATA_PAGE_V2,
1218 uncompressed_page_size: 10,
1219 compressed_page_size: 10,
1220 data_page_header: None,
1221 index_page_header: None,
1222 dictionary_page_header: None,
1223 crc: None,
1224 data_page_header_v2: Some(DataPageHeaderV2 {
1225 num_nulls: 0,
1226 num_rows: 0,
1227 num_values: 0,
1228 encoding: Encoding::PLAIN,
1229 definition_levels_byte_length: 11,
1230 repetition_levels_byte_length: 0,
1231 is_compressed: None,
1232 statistics: None,
1233 }),
1234 };
1235
1236 let buffer = Bytes::new();
1237 let err = decode_page(page_header, buffer, Type::INT32, None).unwrap_err();
1238 assert!(
1239 err.to_string()
1240 .contains("DataPage v2 header contains implausible values")
1241 );
1242 }
1243
1244 #[test]
1245 fn test_decode_unsupported_page() {
1246 let mut page_header = PageHeader {
1247 r#type: PageType::INDEX_PAGE,
1248 uncompressed_page_size: 10,
1249 compressed_page_size: 10,
1250 data_page_header: None,
1251 index_page_header: None,
1252 dictionary_page_header: None,
1253 crc: None,
1254 data_page_header_v2: None,
1255 };
1256 let buffer = Bytes::new();
1257 let err = decode_page(page_header.clone(), buffer.clone(), Type::INT32, None).unwrap_err();
1258 assert_eq!(
1259 err.to_string(),
1260 "Parquet error: Page type INDEX_PAGE is not supported"
1261 );
1262
1263 page_header.data_page_header_v2 = Some(DataPageHeaderV2 {
1264 num_nulls: 0,
1265 num_rows: 0,
1266 num_values: 0,
1267 encoding: Encoding::PLAIN,
1268 definition_levels_byte_length: 11,
1269 repetition_levels_byte_length: 0,
1270 is_compressed: None,
1271 statistics: None,
1272 });
1273 let err = decode_page(page_header, buffer, Type::INT32, None).unwrap_err();
1274 assert!(
1275 err.to_string()
1276 .contains("DataPage v2 header contains implausible values")
1277 );
1278 }
1279
1280 #[test]
1281 fn test_cursor_and_file_has_the_same_behaviour() {
1282 let mut buf: Vec<u8> = Vec::new();
1283 get_test_file("alltypes_plain.parquet")
1284 .read_to_end(&mut buf)
1285 .unwrap();
1286 let cursor = Bytes::from(buf);
1287 let read_from_cursor = SerializedFileReader::new(cursor).unwrap();
1288
1289 let test_file = get_test_file("alltypes_plain.parquet");
1290 let read_from_file = SerializedFileReader::new(test_file).unwrap();
1291
1292 let file_iter = read_from_file.get_row_iter(None).unwrap();
1293 let cursor_iter = read_from_cursor.get_row_iter(None).unwrap();
1294
1295 for (a, b) in file_iter.zip(cursor_iter) {
1296 assert_eq!(a.unwrap(), b.unwrap())
1297 }
1298 }
1299
1300 #[test]
1301 fn test_file_reader_try_from() {
1302 let test_file = get_test_file("alltypes_plain.parquet");
1304 let test_path_buf = get_test_path("alltypes_plain.parquet");
1305 let test_path = test_path_buf.as_path();
1306 let test_path_str = test_path.to_str().unwrap();
1307
1308 let reader = SerializedFileReader::try_from(test_file);
1309 assert!(reader.is_ok());
1310
1311 let reader = SerializedFileReader::try_from(test_path);
1312 assert!(reader.is_ok());
1313
1314 let reader = SerializedFileReader::try_from(test_path_str);
1315 assert!(reader.is_ok());
1316
1317 let reader = SerializedFileReader::try_from(test_path_str.to_string());
1318 assert!(reader.is_ok());
1319
1320 let test_path = Path::new("invalid.parquet");
1322 let test_path_str = test_path.to_str().unwrap();
1323
1324 let reader = SerializedFileReader::try_from(test_path);
1325 assert!(reader.is_err());
1326
1327 let reader = SerializedFileReader::try_from(test_path_str);
1328 assert!(reader.is_err());
1329
1330 let reader = SerializedFileReader::try_from(test_path_str.to_string());
1331 assert!(reader.is_err());
1332 }
1333
1334 #[test]
1335 fn test_file_reader_into_iter() {
1336 let path = get_test_path("alltypes_plain.parquet");
1337 let reader = SerializedFileReader::try_from(path.as_path()).unwrap();
1338 let iter = reader.into_iter();
1339 let values: Vec<_> = iter.flat_map(|x| x.unwrap().get_int(0)).collect();
1340
1341 assert_eq!(values, &[4, 5, 6, 7, 2, 3, 0, 1]);
1342 }
1343
1344 #[test]
1345 fn test_file_reader_into_iter_project() {
1346 let path = get_test_path("alltypes_plain.parquet");
1347 let reader = SerializedFileReader::try_from(path.as_path()).unwrap();
1348 let schema = "message schema { OPTIONAL INT32 id; }";
1349 let proj = parse_message_type(schema).ok();
1350 let iter = reader.into_iter().project(proj).unwrap();
1351 let values: Vec<_> = iter.flat_map(|x| x.unwrap().get_int(0)).collect();
1352
1353 assert_eq!(values, &[4, 5, 6, 7, 2, 3, 0, 1]);
1354 }
1355
1356 #[test]
1357 fn test_reuse_file_chunk() {
1358 let test_file = get_test_file("alltypes_plain.parquet");
1362 let reader = SerializedFileReader::new(test_file).unwrap();
1363 let row_group = reader.get_row_group(0).unwrap();
1364
1365 let mut page_readers = Vec::new();
1366 for i in 0..row_group.num_columns() {
1367 page_readers.push(row_group.get_column_page_reader(i).unwrap());
1368 }
1369
1370 for mut page_reader in page_readers {
1373 assert!(page_reader.get_next_page().is_ok());
1374 }
1375 }
1376
1377 #[test]
1378 fn test_file_reader() {
1379 let test_file = get_test_file("alltypes_plain.parquet");
1380 let reader_result = SerializedFileReader::new(test_file);
1381 assert!(reader_result.is_ok());
1382 let reader = reader_result.unwrap();
1383
1384 let metadata = reader.metadata();
1386 assert_eq!(metadata.num_row_groups(), 1);
1387
1388 let file_metadata = metadata.file_metadata();
1390 assert!(file_metadata.created_by().is_some());
1391 assert_eq!(
1392 file_metadata.created_by().unwrap(),
1393 "impala version 1.3.0-INTERNAL (build 8a48ddb1eff84592b3fc06bc6f51ec120e1fffc9)"
1394 );
1395 assert!(file_metadata.key_value_metadata().is_none());
1396 assert_eq!(file_metadata.num_rows(), 8);
1397 assert_eq!(file_metadata.version(), 1);
1398 assert_eq!(file_metadata.column_orders(), None);
1399
1400 let row_group_metadata = metadata.row_group(0);
1402 assert_eq!(row_group_metadata.num_columns(), 11);
1403 assert_eq!(row_group_metadata.num_rows(), 8);
1404 assert_eq!(row_group_metadata.total_byte_size(), 671);
1405 for i in 0..row_group_metadata.num_columns() {
1407 assert_eq!(file_metadata.column_order(i), ColumnOrder::UNDEFINED);
1408 }
1409
1410 let row_group_reader_result = reader.get_row_group(0);
1412 assert!(row_group_reader_result.is_ok());
1413 let row_group_reader: Box<dyn RowGroupReader> = row_group_reader_result.unwrap();
1414 assert_eq!(
1415 row_group_reader.num_columns(),
1416 row_group_metadata.num_columns()
1417 );
1418 assert_eq!(
1419 row_group_reader.metadata().total_byte_size(),
1420 row_group_metadata.total_byte_size()
1421 );
1422
1423 let page_reader_0_result = row_group_reader.get_column_page_reader(0);
1426 assert!(page_reader_0_result.is_ok());
1427 let mut page_reader_0: Box<dyn PageReader> = page_reader_0_result.unwrap();
1428 let mut page_count = 0;
1429 while let Some(page) = page_reader_0.get_next_page().unwrap() {
1430 let is_expected_page = match page {
1431 Page::DictionaryPage {
1432 buf,
1433 num_values,
1434 encoding,
1435 is_sorted,
1436 } => {
1437 assert_eq!(buf.len(), 32);
1438 assert_eq!(num_values, 8);
1439 assert_eq!(encoding, Encoding::PLAIN_DICTIONARY);
1440 assert!(!is_sorted);
1441 true
1442 }
1443 Page::DataPage {
1444 buf,
1445 num_values,
1446 encoding,
1447 def_level_encoding,
1448 rep_level_encoding,
1449 statistics,
1450 } => {
1451 assert_eq!(buf.len(), 11);
1452 assert_eq!(num_values, 8);
1453 assert_eq!(encoding, Encoding::PLAIN_DICTIONARY);
1454 assert_eq!(def_level_encoding, Encoding::RLE);
1455 #[expect(deprecated)]
1456 let expected_rep_level_encoding = Encoding::BIT_PACKED;
1457 assert_eq!(rep_level_encoding, expected_rep_level_encoding);
1458 assert!(statistics.is_none());
1459 true
1460 }
1461 Page::DataPageV2 { .. } => false,
1462 };
1463 assert!(is_expected_page);
1464 page_count += 1;
1465 }
1466 assert_eq!(page_count, 2);
1467 }
1468
1469 #[test]
1470 fn test_file_reader_datapage_v2() {
1471 let test_file = get_test_file("datapage_v2.snappy.parquet");
1472 let reader_result = SerializedFileReader::new(test_file);
1473 assert!(reader_result.is_ok());
1474 let reader = reader_result.unwrap();
1475
1476 let metadata = reader.metadata();
1478 assert_eq!(metadata.num_row_groups(), 1);
1479
1480 let file_metadata = metadata.file_metadata();
1482 assert!(file_metadata.created_by().is_some());
1483 assert_eq!(
1484 file_metadata.created_by().unwrap(),
1485 "parquet-mr version 1.8.1 (build 4aba4dae7bb0d4edbcf7923ae1339f28fd3f7fcf)"
1486 );
1487 assert!(file_metadata.key_value_metadata().is_some());
1488 assert_eq!(file_metadata.key_value_metadata().unwrap().len(), 1);
1489
1490 assert_eq!(file_metadata.num_rows(), 5);
1491 assert_eq!(file_metadata.version(), 1);
1492 assert_eq!(file_metadata.column_orders(), None);
1493
1494 let row_group_metadata = metadata.row_group(0);
1495
1496 for i in 0..row_group_metadata.num_columns() {
1498 assert_eq!(file_metadata.column_order(i), ColumnOrder::UNDEFINED);
1499 }
1500
1501 let row_group_reader_result = reader.get_row_group(0);
1503 assert!(row_group_reader_result.is_ok());
1504 let row_group_reader: Box<dyn RowGroupReader> = row_group_reader_result.unwrap();
1505 assert_eq!(
1506 row_group_reader.num_columns(),
1507 row_group_metadata.num_columns()
1508 );
1509 assert_eq!(
1510 row_group_reader.metadata().total_byte_size(),
1511 row_group_metadata.total_byte_size()
1512 );
1513
1514 let page_reader_0_result = row_group_reader.get_column_page_reader(0);
1517 assert!(page_reader_0_result.is_ok());
1518 let mut page_reader_0: Box<dyn PageReader> = page_reader_0_result.unwrap();
1519 let mut page_count = 0;
1520 while let Some(page) = page_reader_0.get_next_page().unwrap() {
1521 let is_expected_page = match page {
1522 Page::DictionaryPage {
1523 buf,
1524 num_values,
1525 encoding,
1526 is_sorted,
1527 } => {
1528 assert_eq!(buf.len(), 7);
1529 assert_eq!(num_values, 1);
1530 assert_eq!(encoding, Encoding::PLAIN);
1531 assert!(!is_sorted);
1532 true
1533 }
1534 Page::DataPageV2 {
1535 buf,
1536 num_values,
1537 encoding,
1538 num_nulls,
1539 num_rows,
1540 def_levels_byte_len,
1541 rep_levels_byte_len,
1542 is_compressed,
1543 statistics,
1544 } => {
1545 assert_eq!(buf.len(), 4);
1546 assert_eq!(num_values, 5);
1547 assert_eq!(encoding, Encoding::RLE_DICTIONARY);
1548 assert_eq!(num_nulls, 1);
1549 assert_eq!(num_rows, 5);
1550 assert_eq!(def_levels_byte_len, 2);
1551 assert_eq!(rep_levels_byte_len, 0);
1552 assert!(is_compressed);
1553 assert!(statistics.is_none()); true
1555 }
1556 Page::DataPage { .. } => false,
1557 };
1558 assert!(is_expected_page);
1559 page_count += 1;
1560 }
1561 assert_eq!(page_count, 2);
1562 }
1563
1564 #[cfg_attr(miri, ignore)] #[test]
1566 fn test_file_reader_empty_compressed_datapage_v2() {
1567 let test_file = get_test_file("page_v2_empty_compressed.parquet");
1569 let reader_result = SerializedFileReader::new(test_file);
1570 assert!(reader_result.is_ok());
1571 let reader = reader_result.unwrap();
1572
1573 let metadata = reader.metadata();
1575 assert_eq!(metadata.num_row_groups(), 1);
1576
1577 let file_metadata = metadata.file_metadata();
1579 assert!(file_metadata.created_by().is_some());
1580 assert_eq!(
1581 file_metadata.created_by().unwrap(),
1582 "parquet-cpp-arrow version 14.0.2"
1583 );
1584 assert!(file_metadata.key_value_metadata().is_some());
1585 assert_eq!(file_metadata.key_value_metadata().unwrap().len(), 1);
1586
1587 assert_eq!(file_metadata.num_rows(), 10);
1588 assert_eq!(file_metadata.version(), 2);
1589 let expected_order = ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED);
1590 assert_eq!(
1591 file_metadata.column_orders(),
1592 Some(vec![expected_order].as_ref())
1593 );
1594
1595 let row_group_metadata = metadata.row_group(0);
1596
1597 for i in 0..row_group_metadata.num_columns() {
1599 assert_eq!(file_metadata.column_order(i), expected_order);
1600 }
1601
1602 let row_group_reader_result = reader.get_row_group(0);
1604 assert!(row_group_reader_result.is_ok());
1605 let row_group_reader: Box<dyn RowGroupReader> = row_group_reader_result.unwrap();
1606 assert_eq!(
1607 row_group_reader.num_columns(),
1608 row_group_metadata.num_columns()
1609 );
1610 assert_eq!(
1611 row_group_reader.metadata().total_byte_size(),
1612 row_group_metadata.total_byte_size()
1613 );
1614
1615 let page_reader_0_result = row_group_reader.get_column_page_reader(0);
1617 assert!(page_reader_0_result.is_ok());
1618 let mut page_reader_0: Box<dyn PageReader> = page_reader_0_result.unwrap();
1619 let mut page_count = 0;
1620 while let Some(page) = page_reader_0.get_next_page().unwrap() {
1621 let is_expected_page = match page {
1622 Page::DictionaryPage {
1623 buf,
1624 num_values,
1625 encoding,
1626 is_sorted,
1627 } => {
1628 assert_eq!(buf.len(), 0);
1629 assert_eq!(num_values, 0);
1630 assert_eq!(encoding, Encoding::PLAIN);
1631 assert!(!is_sorted);
1632 true
1633 }
1634 Page::DataPageV2 {
1635 buf,
1636 num_values,
1637 encoding,
1638 num_nulls,
1639 num_rows,
1640 def_levels_byte_len,
1641 rep_levels_byte_len,
1642 is_compressed,
1643 statistics,
1644 } => {
1645 assert_eq!(buf.len(), 3);
1646 assert_eq!(num_values, 10);
1647 assert_eq!(encoding, Encoding::RLE_DICTIONARY);
1648 assert_eq!(num_nulls, 10);
1649 assert_eq!(num_rows, 10);
1650 assert_eq!(def_levels_byte_len, 2);
1651 assert_eq!(rep_levels_byte_len, 0);
1652 assert!(is_compressed);
1653 assert!(statistics.is_none()); true
1655 }
1656 Page::DataPage { .. } => false,
1657 };
1658 assert!(is_expected_page);
1659 page_count += 1;
1660 }
1661 assert_eq!(page_count, 2);
1662 }
1663
1664 #[test]
1665 fn test_file_reader_empty_datapage_v2() {
1666 let test_file = get_test_file("datapage_v2_empty_datapage.snappy.parquet");
1668 let reader_result = SerializedFileReader::new(test_file);
1669 assert!(reader_result.is_ok());
1670 let reader = reader_result.unwrap();
1671
1672 let metadata = reader.metadata();
1674 assert_eq!(metadata.num_row_groups(), 1);
1675
1676 let file_metadata = metadata.file_metadata();
1678 assert!(file_metadata.created_by().is_some());
1679 assert_eq!(
1680 file_metadata.created_by().unwrap(),
1681 "parquet-mr version 1.13.1 (build db4183109d5b734ec5930d870cdae161e408ddba)"
1682 );
1683 assert!(file_metadata.key_value_metadata().is_some());
1684 assert_eq!(file_metadata.key_value_metadata().unwrap().len(), 2);
1685
1686 assert_eq!(file_metadata.num_rows(), 1);
1687 assert_eq!(file_metadata.version(), 1);
1688 let expected_order = ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED);
1689 assert_eq!(
1690 file_metadata.column_orders(),
1691 Some(vec![expected_order].as_ref())
1692 );
1693
1694 let row_group_metadata = metadata.row_group(0);
1695
1696 for i in 0..row_group_metadata.num_columns() {
1698 assert_eq!(file_metadata.column_order(i), expected_order);
1699 }
1700
1701 let row_group_reader_result = reader.get_row_group(0);
1703 assert!(row_group_reader_result.is_ok());
1704 let row_group_reader: Box<dyn RowGroupReader> = row_group_reader_result.unwrap();
1705 assert_eq!(
1706 row_group_reader.num_columns(),
1707 row_group_metadata.num_columns()
1708 );
1709 assert_eq!(
1710 row_group_reader.metadata().total_byte_size(),
1711 row_group_metadata.total_byte_size()
1712 );
1713
1714 let page_reader_0_result = row_group_reader.get_column_page_reader(0);
1716 assert!(page_reader_0_result.is_ok());
1717 let mut page_reader_0: Box<dyn PageReader> = page_reader_0_result.unwrap();
1718 let mut page_count = 0;
1719 while let Some(page) = page_reader_0.get_next_page().unwrap() {
1720 let is_expected_page = match page {
1721 Page::DataPageV2 {
1722 buf,
1723 num_values,
1724 encoding,
1725 num_nulls,
1726 num_rows,
1727 def_levels_byte_len,
1728 rep_levels_byte_len,
1729 is_compressed,
1730 statistics,
1731 } => {
1732 assert_eq!(buf.len(), 2);
1733 assert_eq!(num_values, 1);
1734 assert_eq!(encoding, Encoding::PLAIN);
1735 assert_eq!(num_nulls, 1);
1736 assert_eq!(num_rows, 1);
1737 assert_eq!(def_levels_byte_len, 2);
1738 assert_eq!(rep_levels_byte_len, 0);
1739 assert!(is_compressed);
1740 assert!(statistics.is_none());
1741 true
1742 }
1743 _ => false,
1744 };
1745 assert!(is_expected_page);
1746 page_count += 1;
1747 }
1748 assert_eq!(page_count, 1);
1749 }
1750
1751 fn get_serialized_page_reader<R: ChunkReader>(
1752 file_reader: &SerializedFileReader<R>,
1753 row_group_idx: usize,
1754 column: usize,
1755 ) -> Result<SerializedPageReader<R>> {
1756 let row_group = {
1757 let row_group_metadata = file_reader.metadata.row_group(row_group_idx);
1758 let props = Arc::clone(&file_reader.props);
1759 let f = Arc::clone(&file_reader.chunk_reader);
1760 let page_index = file_reader.metadata.page_index_for_row_group(row_group_idx);
1761 SerializedRowGroupReader::new(f, row_group_metadata, page_index, props)?
1762 };
1763
1764 let col = row_group.metadata.column(column);
1765 let page_locations = file_reader
1766 .metadata
1767 .page_index()
1768 .map(|pi| pi.page_locations(row_group_idx, column))
1769 .unwrap_or(None)
1770 .cloned();
1771
1772 let props = Arc::clone(&row_group.props);
1773 SerializedPageReader::new_with_properties(
1774 Arc::clone(&row_group.chunk_reader),
1775 col,
1776 usize::try_from(row_group.metadata.num_rows())?,
1777 page_locations,
1778 props,
1779 )
1780 }
1781
1782 #[test]
1783 fn test_peek_next_page_offset_matches_actual() -> Result<()> {
1784 let test_file = get_test_file("alltypes_plain.parquet");
1785 let reader = SerializedFileReader::new(test_file)?;
1786
1787 let mut offset_set = HashSet::new();
1788 let num_row_groups = reader.metadata.num_row_groups();
1789 for row_group in 0..num_row_groups {
1790 let num_columns = reader.metadata.row_group(row_group).num_columns();
1791 for column in 0..num_columns {
1792 let mut page_reader = get_serialized_page_reader(&reader, row_group, column)?;
1793
1794 while let Ok(Some(page_offset)) = page_reader.peek_next_page_offset() {
1795 match &page_reader.state {
1796 SerializedPageReaderState::Pages {
1797 page_locations,
1798 dictionary_page,
1799 ..
1800 } => {
1801 if let Some(page) = dictionary_page {
1802 assert_eq!(page.offset as u64, page_offset);
1803 } else if let Some(page) = page_locations.front() {
1804 assert_eq!(page.offset as u64, page_offset);
1805 } else {
1806 unreachable!()
1807 }
1808 }
1809 SerializedPageReaderState::Values {
1810 offset,
1811 next_page_header,
1812 ..
1813 } => {
1814 assert!(next_page_header.is_some());
1815 assert_eq!(*offset, page_offset);
1816 }
1817 }
1818 let page = page_reader.get_next_page()?;
1819 assert!(page.is_some());
1820 let newly_inserted = offset_set.insert(page_offset);
1821 assert!(newly_inserted);
1822 }
1823 }
1824 }
1825
1826 Ok(())
1827 }
1828
1829 #[test]
1830 fn test_page_iterator() {
1831 let file = get_test_file("alltypes_plain.parquet");
1832 let file_reader = Arc::new(SerializedFileReader::new(file).unwrap());
1833
1834 let mut page_iterator = FilePageIterator::new(0, file_reader.clone()).unwrap();
1835
1836 let page = page_iterator.next();
1838 assert!(page.is_some());
1839 assert!(page.unwrap().is_ok());
1840
1841 let page = page_iterator.next();
1843 assert!(page.is_none());
1844
1845 let row_group_indices = Box::new(0..1);
1846 let mut page_iterator =
1847 FilePageIterator::with_row_groups(0, row_group_indices, file_reader).unwrap();
1848
1849 let page = page_iterator.next();
1851 assert!(page.is_some());
1852 assert!(page.unwrap().is_ok());
1853
1854 let page = page_iterator.next();
1856 assert!(page.is_none());
1857 }
1858
1859 #[test]
1860 fn test_file_reader_key_value_metadata() {
1861 let file = get_test_file("binary.parquet");
1862 let file_reader = Arc::new(SerializedFileReader::new(file).unwrap());
1863
1864 let metadata = file_reader
1865 .metadata
1866 .file_metadata()
1867 .key_value_metadata()
1868 .unwrap();
1869
1870 assert_eq!(metadata.len(), 3);
1871
1872 assert_eq!(metadata[0].key, "parquet.proto.descriptor");
1873
1874 assert_eq!(metadata[1].key, "writer.model.name");
1875 assert_eq!(metadata[1].value, Some("protobuf".to_owned()));
1876
1877 assert_eq!(metadata[2].key, "parquet.proto.class");
1878 assert_eq!(metadata[2].value, Some("foo.baz.Foobaz$Event".to_owned()));
1879 }
1880
1881 #[test]
1882 fn test_file_reader_optional_metadata() {
1883 let file = get_test_file("data_index_bloom_encoding_stats.parquet");
1885 let options = ReadOptionsBuilder::new()
1886 .with_encoding_stats_as_mask(false)
1887 .build();
1888 let file_reader = Arc::new(SerializedFileReader::new_with_options(file, options).unwrap());
1889
1890 let row_group_metadata = file_reader.metadata.row_group(0);
1891 let col0_metadata = row_group_metadata.column(0);
1892
1893 assert_eq!(col0_metadata.bloom_filter_offset().unwrap(), 192);
1895
1896 let page_encoding_stats = &col0_metadata.page_encoding_stats().unwrap()[0];
1898
1899 assert_eq!(page_encoding_stats.page_type, basic::PageType::DATA_PAGE);
1900 assert_eq!(page_encoding_stats.encoding, Encoding::PLAIN);
1901 assert_eq!(page_encoding_stats.count, 1);
1902
1903 assert_eq!(col0_metadata.column_index_offset().unwrap(), 156);
1905 assert_eq!(col0_metadata.column_index_length().unwrap(), 25);
1906
1907 assert_eq!(col0_metadata.offset_index_offset().unwrap(), 181);
1909 assert_eq!(col0_metadata.offset_index_length().unwrap(), 11);
1910 }
1911
1912 #[test]
1913 fn test_file_reader_page_stats_mask() {
1914 let file = get_test_file("alltypes_tiny_pages.parquet");
1915 let options = ReadOptionsBuilder::new()
1916 .with_encoding_stats_as_mask(true)
1917 .build();
1918 let file_reader = Arc::new(SerializedFileReader::new_with_options(file, options).unwrap());
1919
1920 let row_group_metadata = file_reader.metadata.row_group(0);
1921
1922 let page_encoding_stats = row_group_metadata
1924 .column(0)
1925 .page_encoding_stats_mask()
1926 .unwrap();
1927 assert!(page_encoding_stats.is_only(Encoding::PLAIN));
1928 let page_encoding_stats = row_group_metadata
1929 .column(2)
1930 .page_encoding_stats_mask()
1931 .unwrap();
1932 assert!(page_encoding_stats.is_only(Encoding::PLAIN_DICTIONARY));
1933 }
1934
1935 #[test]
1936 fn test_file_reader_page_stats_skipped() {
1937 let file = get_test_file("alltypes_tiny_pages.parquet");
1938
1939 let options = ReadOptionsBuilder::new()
1941 .with_encoding_stats_policy(ParquetStatisticsPolicy::SkipAll)
1942 .with_column_stats_policy(ParquetStatisticsPolicy::SkipAll)
1943 .build();
1944 let file_reader = Arc::new(
1945 SerializedFileReader::new_with_options(file.try_clone().unwrap(), options).unwrap(),
1946 );
1947
1948 let row_group_metadata = file_reader.metadata.row_group(0);
1949 for column in row_group_metadata.columns() {
1950 assert!(column.page_encoding_stats().is_none());
1951 assert!(column.page_encoding_stats_mask().is_none());
1952 assert!(column.statistics().is_none());
1953 }
1954
1955 let options = ReadOptionsBuilder::new()
1957 .with_encoding_stats_as_mask(true)
1958 .with_encoding_stats_policy(ParquetStatisticsPolicy::skip_except(&[0]))
1959 .with_column_stats_policy(ParquetStatisticsPolicy::skip_except(&[0]))
1960 .build();
1961 let file_reader = Arc::new(
1962 SerializedFileReader::new_with_options(file.try_clone().unwrap(), options).unwrap(),
1963 );
1964
1965 let row_group_metadata = file_reader.metadata.row_group(0);
1966 for (idx, column) in row_group_metadata.columns().iter().enumerate() {
1967 assert!(column.page_encoding_stats().is_none());
1968 assert_eq!(column.page_encoding_stats_mask().is_some(), idx == 0);
1969 assert_eq!(column.statistics().is_some(), idx == 0);
1970 }
1971 }
1972
1973 #[test]
1974 fn test_file_reader_size_stats_skipped() {
1975 let file = get_test_file("repeated_primitive_no_list.parquet");
1976
1977 let options = ReadOptionsBuilder::new()
1979 .with_size_stats_policy(ParquetStatisticsPolicy::SkipAll)
1980 .build();
1981 let file_reader = Arc::new(
1982 SerializedFileReader::new_with_options(file.try_clone().unwrap(), options).unwrap(),
1983 );
1984
1985 let row_group_metadata = file_reader.metadata.row_group(0);
1986 for column in row_group_metadata.columns() {
1987 assert!(column.repetition_level_histogram().is_none());
1988 assert!(column.definition_level_histogram().is_none());
1989 assert!(column.unencoded_byte_array_data_bytes().is_none());
1990 }
1991
1992 let options = ReadOptionsBuilder::new()
1994 .with_encoding_stats_as_mask(true)
1995 .with_size_stats_policy(ParquetStatisticsPolicy::skip_except(&[1]))
1996 .build();
1997 let file_reader = Arc::new(
1998 SerializedFileReader::new_with_options(file.try_clone().unwrap(), options).unwrap(),
1999 );
2000
2001 let row_group_metadata = file_reader.metadata.row_group(0);
2002 for (idx, column) in row_group_metadata.columns().iter().enumerate() {
2003 assert_eq!(column.repetition_level_histogram().is_some(), idx == 1);
2004 assert_eq!(column.definition_level_histogram().is_some(), idx == 1);
2005 assert_eq!(column.unencoded_byte_array_data_bytes().is_some(), idx == 1);
2006 }
2007 }
2008
2009 #[test]
2010 fn test_file_reader_with_no_filter() -> Result<()> {
2011 let test_file = get_test_file("alltypes_plain.parquet");
2012 let origin_reader = SerializedFileReader::new(test_file)?;
2013 let metadata = origin_reader.metadata();
2015 assert_eq!(metadata.num_row_groups(), 1);
2016 Ok(())
2017 }
2018
2019 #[test]
2020 fn test_file_reader_filter_row_groups_with_predicate() -> Result<()> {
2021 let test_file = get_test_file("alltypes_plain.parquet");
2022 let read_options = ReadOptionsBuilder::new()
2023 .with_predicate(Box::new(|_, _| false))
2024 .build();
2025 let reader = SerializedFileReader::new_with_options(test_file, read_options)?;
2026 let metadata = reader.metadata();
2027 assert_eq!(metadata.num_row_groups(), 0);
2028 Ok(())
2029 }
2030
2031 #[test]
2032 fn test_file_reader_filter_row_groups_with_range() -> Result<()> {
2033 let test_file = get_test_file("alltypes_plain.parquet");
2034 let origin_reader = SerializedFileReader::new(test_file)?;
2035 let metadata = origin_reader.metadata();
2037 assert_eq!(metadata.num_row_groups(), 1);
2038 let mid = get_midpoint_offset(metadata.row_group(0));
2039
2040 let test_file = get_test_file("alltypes_plain.parquet");
2041 let read_options = ReadOptionsBuilder::new().with_range(0, mid + 1).build();
2042 let reader = SerializedFileReader::new_with_options(test_file, read_options)?;
2043 let metadata = reader.metadata();
2044 assert_eq!(metadata.num_row_groups(), 1);
2045
2046 let test_file = get_test_file("alltypes_plain.parquet");
2047 let read_options = ReadOptionsBuilder::new().with_range(0, mid).build();
2048 let reader = SerializedFileReader::new_with_options(test_file, read_options)?;
2049 let metadata = reader.metadata();
2050 assert_eq!(metadata.num_row_groups(), 0);
2051 Ok(())
2052 }
2053
2054 #[test]
2055 #[cfg_attr(miri, ignore)] fn test_file_reader_filter_row_groups_and_range() -> Result<()> {
2057 let test_file = get_test_file("alltypes_tiny_pages.parquet");
2058 let origin_reader = SerializedFileReader::new(test_file)?;
2059 let metadata = origin_reader.metadata();
2060 let mid = get_midpoint_offset(metadata.row_group(0));
2061
2062 let test_file = get_test_file("alltypes_tiny_pages.parquet");
2064 let read_options = ReadOptionsBuilder::new()
2065 .with_page_index()
2066 .with_predicate(Box::new(|_, _| true))
2067 .with_range(mid, mid + 1)
2068 .build();
2069 let reader = SerializedFileReader::new_with_options(test_file, read_options)?;
2070 let metadata = reader.metadata();
2071 assert_eq!(metadata.num_row_groups(), 1);
2072 assert!(metadata.page_index().is_some());
2073
2074 let test_file = get_test_file("alltypes_tiny_pages.parquet");
2076 let read_options = ReadOptionsBuilder::new()
2077 .with_page_index()
2078 .with_predicate(Box::new(|_, _| true))
2079 .with_range(0, mid)
2080 .build();
2081 let reader = SerializedFileReader::new_with_options(test_file, read_options)?;
2082 let metadata = reader.metadata();
2083 assert_eq!(metadata.num_row_groups(), 0);
2084 assert!(metadata.page_index().is_none());
2085
2086 let test_file = get_test_file("alltypes_tiny_pages.parquet");
2088 let read_options = ReadOptionsBuilder::new()
2089 .with_page_index()
2090 .with_predicate(Box::new(|_, _| false))
2091 .with_range(mid, mid + 1)
2092 .build();
2093 let reader = SerializedFileReader::new_with_options(test_file, read_options)?;
2094 let metadata = reader.metadata();
2095 assert_eq!(metadata.num_row_groups(), 0);
2096 assert!(metadata.page_index().is_none());
2097
2098 let test_file = get_test_file("alltypes_tiny_pages.parquet");
2100 let read_options = ReadOptionsBuilder::new()
2101 .with_page_index()
2102 .with_predicate(Box::new(|_, _| false))
2103 .with_range(0, mid)
2104 .build();
2105 let reader = SerializedFileReader::new_with_options(test_file, read_options)?;
2106 let metadata = reader.metadata();
2107 assert_eq!(metadata.num_row_groups(), 0);
2108 assert!(metadata.page_index().is_none());
2109 Ok(())
2110 }
2111
2112 #[test]
2113 fn test_file_reader_invalid_metadata() {
2114 let data = [
2115 255, 172, 1, 0, 50, 82, 65, 73, 1, 0, 0, 0, 169, 168, 168, 162, 87, 255, 16, 0, 0, 0,
2116 80, 65, 82, 49,
2117 ];
2118 let ret = SerializedFileReader::new(Bytes::copy_from_slice(&data));
2119 assert_eq!(
2120 ret.err().unwrap().to_string(),
2121 "Parquet error: Expected list element type of Struct but got List"
2122 );
2123 }
2124
2125 #[test]
2126 fn test_page_index_reader() {
2143 let test_file = get_test_file("data_index_bloom_encoding_stats.parquet");
2144 let builder = ReadOptionsBuilder::new();
2145 let options = builder.with_page_index().build();
2147 let reader_result = SerializedFileReader::new_with_options(test_file, options);
2148 let reader = reader_result.unwrap();
2149
2150 let metadata = reader.metadata();
2152 assert_eq!(metadata.num_row_groups(), 1);
2153
2154 let page_index = metadata.page_index().expect("page index should be present");
2155
2156 let Some(ColumnIndexMetaData::BYTE_ARRAY(index)) = page_index.column_index(0, 0) else {
2158 unreachable!()
2159 };
2160
2161 assert_eq!(index.boundary_order, BoundaryOrder::ASCENDING);
2162
2163 assert_eq!(index.num_pages(), 1);
2165
2166 let min = index.min_value(0).unwrap();
2167 let max = index.max_value(0).unwrap();
2168 assert_eq!(b"Hello", min.as_bytes());
2169 assert_eq!(b"today", max.as_bytes());
2170
2171 let offset_index = page_index
2173 .offset_index(0, 0)
2174 .expect("offset index should be present");
2175 let page_offset = offset_index
2176 .page_locations()
2177 .first()
2178 .expect("offset index too small");
2179
2180 assert_eq!(4, page_offset.offset);
2181 assert_eq!(152, page_offset.compressed_page_size);
2182 assert_eq!(0, page_offset.first_row_index);
2183 }
2184
2185 #[test]
2186 #[cfg_attr(miri, ignore)] fn test_page_index_reader_all_type() {
2188 let test_file = get_test_file("alltypes_tiny_pages_plain.parquet");
2189 let builder = ReadOptionsBuilder::new();
2190 let options = builder.with_page_index().build();
2192 let reader_result = SerializedFileReader::new_with_options(test_file, options);
2193 let reader = reader_result.unwrap();
2194
2195 let metadata = reader.metadata();
2197 assert_eq!(metadata.num_row_groups(), 1);
2198
2199 let page_index = metadata.page_index().unwrap();
2200 let row_group_offset_indexes = metadata.page_index_for_row_group(0);
2201
2202 let row_group_metadata = metadata.row_group(0);
2204
2205 let ci = page_index.column_index(0, 0).unwrap();
2207 assert!(!ci.is_sorted());
2208 assert!(matches!(
2209 ci.get_boundary_order(),
2210 Some(BoundaryOrder::UNORDERED)
2211 ));
2212 if let ColumnIndexMetaData::INT32(index) = ci {
2213 check_native_page_index(
2214 index,
2215 325,
2216 get_row_group_min_max_bytes(row_group_metadata, 0),
2217 BoundaryOrder::UNORDERED,
2218 );
2219 assert_eq!(
2220 row_group_offset_indexes.page_locations(0).unwrap().len(),
2221 325
2222 );
2223 } else {
2224 unreachable!()
2225 }
2226 let ci = page_index.column_index(0, 1).unwrap();
2228 assert!(ci.is_sorted());
2229 if let ColumnIndexMetaData::BOOLEAN(index) = ci {
2230 assert_eq!(index.num_pages(), 82);
2231 assert_eq!(
2232 row_group_offset_indexes.page_locations(1).unwrap().len(),
2233 82
2234 );
2235 } else {
2236 unreachable!()
2237 }
2238 let ci = page_index.column_index(0, 2).unwrap();
2240 assert!(ci.is_sorted());
2241 if let ColumnIndexMetaData::INT32(index) = ci {
2242 check_native_page_index(
2243 index,
2244 325,
2245 get_row_group_min_max_bytes(row_group_metadata, 2),
2246 BoundaryOrder::ASCENDING,
2247 );
2248 assert_eq!(
2249 row_group_offset_indexes.page_locations(2).unwrap().len(),
2250 325
2251 );
2252 } else {
2253 unreachable!()
2254 }
2255 let ci = page_index.column_index(0, 3).unwrap();
2257 assert!(ci.is_sorted());
2258 if let ColumnIndexMetaData::INT32(index) = ci {
2259 check_native_page_index(
2260 index,
2261 325,
2262 get_row_group_min_max_bytes(row_group_metadata, 3),
2263 BoundaryOrder::ASCENDING,
2264 );
2265 assert_eq!(
2266 row_group_offset_indexes.page_locations(3).unwrap().len(),
2267 325
2268 );
2269 } else {
2270 unreachable!()
2271 }
2272 let ci = page_index.column_index(0, 4).unwrap();
2274 assert!(ci.is_sorted());
2275 if let ColumnIndexMetaData::INT32(index) = ci {
2276 check_native_page_index(
2277 index,
2278 325,
2279 get_row_group_min_max_bytes(row_group_metadata, 4),
2280 BoundaryOrder::ASCENDING,
2281 );
2282 assert_eq!(
2283 row_group_offset_indexes.page_locations(4).unwrap().len(),
2284 325
2285 );
2286 } else {
2287 unreachable!()
2288 }
2289 let ci = page_index.column_index(0, 5).unwrap();
2291 assert!(!ci.is_sorted());
2292 if let ColumnIndexMetaData::INT64(index) = ci {
2293 check_native_page_index(
2294 index,
2295 528,
2296 get_row_group_min_max_bytes(row_group_metadata, 5),
2297 BoundaryOrder::UNORDERED,
2298 );
2299 assert_eq!(
2300 row_group_offset_indexes.page_locations(5).unwrap().len(),
2301 528
2302 );
2303 } else {
2304 unreachable!()
2305 }
2306 let ci = page_index.column_index(0, 6).unwrap();
2308 assert!(ci.is_sorted());
2309 if let ColumnIndexMetaData::FLOAT(index) = ci {
2310 check_native_page_index(
2311 index,
2312 325,
2313 get_row_group_min_max_bytes(row_group_metadata, 6),
2314 BoundaryOrder::ASCENDING,
2315 );
2316 assert_eq!(
2317 row_group_offset_indexes.page_locations(6).unwrap().len(),
2318 325
2319 );
2320 } else {
2321 unreachable!()
2322 }
2323 let ci = page_index.column_index(0, 7).unwrap();
2325 assert!(!ci.is_sorted());
2326 if let ColumnIndexMetaData::DOUBLE(index) = ci {
2327 check_native_page_index(
2328 index,
2329 528,
2330 get_row_group_min_max_bytes(row_group_metadata, 7),
2331 BoundaryOrder::UNORDERED,
2332 );
2333 assert_eq!(
2334 row_group_offset_indexes.page_locations(7).unwrap().len(),
2335 528
2336 );
2337 } else {
2338 unreachable!()
2339 }
2340 let ci = page_index.column_index(0, 8).unwrap();
2342 assert!(!ci.is_sorted());
2343 if let ColumnIndexMetaData::BYTE_ARRAY(index) = ci {
2344 check_byte_array_page_index(
2345 index,
2346 974,
2347 get_row_group_min_max_bytes(row_group_metadata, 8),
2348 BoundaryOrder::UNORDERED,
2349 );
2350 assert_eq!(
2351 row_group_offset_indexes.page_locations(8).unwrap().len(),
2352 974
2353 );
2354 } else {
2355 unreachable!()
2356 }
2357 let ci = page_index.column_index(0, 9).unwrap();
2359 assert!(ci.is_sorted());
2360 if let ColumnIndexMetaData::BYTE_ARRAY(index) = ci {
2361 check_byte_array_page_index(
2362 index,
2363 352,
2364 get_row_group_min_max_bytes(row_group_metadata, 9),
2365 BoundaryOrder::ASCENDING,
2366 );
2367 assert_eq!(
2368 row_group_offset_indexes.page_locations(9).unwrap().len(),
2369 352
2370 );
2371 } else {
2372 unreachable!()
2373 }
2374 assert!(page_index.column_index(0, 10).is_none());
2377 let ci = page_index.column_index(0, 11).unwrap();
2379 assert!(ci.is_sorted());
2380 if let ColumnIndexMetaData::INT32(index) = ci {
2381 check_native_page_index(
2382 index,
2383 325,
2384 get_row_group_min_max_bytes(row_group_metadata, 11),
2385 BoundaryOrder::ASCENDING,
2386 );
2387 assert_eq!(
2388 row_group_offset_indexes.page_locations(11).unwrap().len(),
2389 325
2390 );
2391 } else {
2392 unreachable!()
2393 }
2394 let ci = page_index.column_index(0, 12).unwrap();
2396 assert!(!ci.is_sorted());
2397 if let ColumnIndexMetaData::INT32(index) = ci {
2398 check_native_page_index(
2399 index,
2400 325,
2401 get_row_group_min_max_bytes(row_group_metadata, 12),
2402 BoundaryOrder::UNORDERED,
2403 );
2404 assert_eq!(
2405 row_group_offset_indexes.page_locations(12).unwrap().len(),
2406 325
2407 );
2408 } else {
2409 unreachable!()
2410 }
2411 }
2412
2413 fn check_native_page_index<T: ParquetValueType>(
2414 row_group_index: &PrimitiveColumnIndex<T>,
2415 page_size: usize,
2416 min_max: (&[u8], &[u8]),
2417 boundary_order: BoundaryOrder,
2418 ) {
2419 assert_eq!(row_group_index.num_pages() as usize, page_size);
2420 assert_eq!(row_group_index.boundary_order, boundary_order);
2421 assert!(row_group_index.min_values().iter().all(|x| {
2422 x >= &T::try_from_le_slice(min_max.0).unwrap()
2423 && x <= &T::try_from_le_slice(min_max.1).unwrap()
2424 }));
2425 }
2426
2427 fn check_byte_array_page_index(
2428 row_group_index: &ByteArrayColumnIndex,
2429 page_size: usize,
2430 min_max: (&[u8], &[u8]),
2431 boundary_order: BoundaryOrder,
2432 ) {
2433 assert_eq!(row_group_index.num_pages() as usize, page_size);
2434 assert_eq!(row_group_index.boundary_order, boundary_order);
2435 for i in 0..row_group_index.num_pages() as usize {
2436 let x = row_group_index.min_value(i).unwrap();
2437 assert!(x >= min_max.0 && x <= min_max.1);
2438 }
2439 }
2440
2441 fn get_row_group_min_max_bytes(r: &RowGroupMetaData, col_num: usize) -> (&[u8], &[u8]) {
2442 let statistics = r.column(col_num).statistics().unwrap();
2443 (
2444 statistics.min_bytes_opt().unwrap_or_default(),
2445 statistics.max_bytes_opt().unwrap_or_default(),
2446 )
2447 }
2448
2449 #[test]
2450 #[cfg_attr(miri, ignore)] fn test_skip_next_page_with_dictionary_page() {
2452 let test_file = get_test_file("alltypes_tiny_pages.parquet");
2453 let builder = ReadOptionsBuilder::new();
2454 let options = builder.with_page_index().build();
2456 let reader_result = SerializedFileReader::new_with_options(test_file, options);
2457 let reader = reader_result.unwrap();
2458
2459 let row_group_reader = reader.get_row_group(0).unwrap();
2460
2461 let mut column_page_reader = row_group_reader.get_column_page_reader(9).unwrap();
2463
2464 let mut vec = vec![];
2465
2466 let meta = column_page_reader.peek_next_page().unwrap().unwrap();
2468 assert!(meta.is_dict);
2469
2470 column_page_reader.skip_next_page().unwrap();
2472
2473 let page = column_page_reader.get_next_page().unwrap().unwrap();
2475 assert!(matches!(page.page_type(), basic::PageType::DATA_PAGE));
2476
2477 for _i in 0..351 {
2479 let meta = column_page_reader.peek_next_page().unwrap().unwrap();
2481 assert!(!meta.is_dict); vec.push(meta);
2483
2484 let page = column_page_reader.get_next_page().unwrap().unwrap();
2485 assert!(matches!(page.page_type(), basic::PageType::DATA_PAGE));
2486 }
2487
2488 assert!(column_page_reader.peek_next_page().unwrap().is_none());
2490 assert!(column_page_reader.get_next_page().unwrap().is_none());
2491
2492 assert_eq!(vec.len(), 351);
2494 }
2495
2496 #[test]
2497 #[cfg_attr(miri, ignore)] fn test_skip_page_with_offset_index() {
2499 let test_file = get_test_file("alltypes_tiny_pages_plain.parquet");
2500 let builder = ReadOptionsBuilder::new();
2501 let options = builder.with_page_index().build();
2503 let reader_result = SerializedFileReader::new_with_options(test_file, options);
2504 let reader = reader_result.unwrap();
2505
2506 let row_group_reader = reader.get_row_group(0).unwrap();
2507
2508 let mut column_page_reader = row_group_reader.get_column_page_reader(4).unwrap();
2510
2511 let mut vec = vec![];
2512
2513 for i in 0..325 {
2514 if i % 2 == 0 {
2515 vec.push(column_page_reader.get_next_page().unwrap().unwrap());
2516 } else {
2517 column_page_reader.skip_next_page().unwrap();
2518 }
2519 }
2520 assert!(column_page_reader.peek_next_page().unwrap().is_none());
2522 assert!(column_page_reader.get_next_page().unwrap().is_none());
2523
2524 assert_eq!(vec.len(), 163);
2525 }
2526
2527 #[test]
2528 fn test_skip_page_without_offset_index() {
2529 let test_file = get_test_file("alltypes_tiny_pages_plain.parquet");
2530
2531 let reader_result = SerializedFileReader::new(test_file);
2533 let reader = reader_result.unwrap();
2534
2535 let row_group_reader = reader.get_row_group(0).unwrap();
2536
2537 let mut column_page_reader = row_group_reader.get_column_page_reader(4).unwrap();
2539
2540 let mut vec = vec![];
2541
2542 for i in 0..325 {
2543 if i % 2 == 0 {
2544 vec.push(column_page_reader.get_next_page().unwrap().unwrap());
2545 } else {
2546 column_page_reader.peek_next_page().unwrap().unwrap();
2547 column_page_reader.skip_next_page().unwrap();
2548 }
2549 }
2550 assert!(column_page_reader.peek_next_page().unwrap().is_none());
2552 assert!(column_page_reader.get_next_page().unwrap().is_none());
2553
2554 assert_eq!(vec.len(), 163);
2555 }
2556
2557 #[test]
2558 #[cfg_attr(miri, ignore)] fn test_peek_page_with_dictionary_page() {
2560 let test_file = get_test_file("alltypes_tiny_pages.parquet");
2561 let builder = ReadOptionsBuilder::new();
2562 let options = builder.with_page_index().build();
2564 let reader_result = SerializedFileReader::new_with_options(test_file, options);
2565 let reader = reader_result.unwrap();
2566 let row_group_reader = reader.get_row_group(0).unwrap();
2567
2568 let mut column_page_reader = row_group_reader.get_column_page_reader(9).unwrap();
2570
2571 let mut vec = vec![];
2572
2573 let meta = column_page_reader.peek_next_page().unwrap().unwrap();
2574 assert!(meta.is_dict);
2575 let page = column_page_reader.get_next_page().unwrap().unwrap();
2576 assert!(matches!(page.page_type(), basic::PageType::DICTIONARY_PAGE));
2577
2578 for i in 0..352 {
2579 let meta = column_page_reader.peek_next_page().unwrap().unwrap();
2580 if i != 351 {
2583 assert!((meta.num_rows == Some(21)) || (meta.num_rows == Some(20)));
2584 } else {
2585 assert_eq!(meta.num_rows, Some(10));
2588 }
2589 assert!(!meta.is_dict);
2590 vec.push(meta);
2591 let page = column_page_reader.get_next_page().unwrap().unwrap();
2592 assert!(matches!(page.page_type(), basic::PageType::DATA_PAGE));
2593 }
2594
2595 assert!(column_page_reader.peek_next_page().unwrap().is_none());
2597 assert!(column_page_reader.get_next_page().unwrap().is_none());
2598
2599 assert_eq!(vec.len(), 352);
2600 }
2601
2602 #[test]
2603 fn test_peek_page_with_dictionary_page_without_offset_index() {
2604 let test_file = get_test_file("alltypes_tiny_pages.parquet");
2605
2606 let reader_result = SerializedFileReader::new(test_file);
2607 let reader = reader_result.unwrap();
2608 let row_group_reader = reader.get_row_group(0).unwrap();
2609
2610 let mut column_page_reader = row_group_reader.get_column_page_reader(9).unwrap();
2612
2613 let mut vec = vec![];
2614
2615 let meta = column_page_reader.peek_next_page().unwrap().unwrap();
2616 assert!(meta.is_dict);
2617 let page = column_page_reader.get_next_page().unwrap().unwrap();
2618 assert!(matches!(page.page_type(), basic::PageType::DICTIONARY_PAGE));
2619
2620 for i in 0..352 {
2621 let meta = column_page_reader.peek_next_page().unwrap().unwrap();
2622 if i != 351 {
2625 assert!((meta.num_levels == Some(21)) || (meta.num_levels == Some(20)));
2626 } else {
2627 assert_eq!(meta.num_levels, Some(10));
2630 }
2631 assert!(!meta.is_dict);
2632 vec.push(meta);
2633 let page = column_page_reader.get_next_page().unwrap().unwrap();
2634 assert!(matches!(page.page_type(), basic::PageType::DATA_PAGE));
2635 }
2636
2637 assert!(column_page_reader.peek_next_page().unwrap().is_none());
2639 assert!(column_page_reader.get_next_page().unwrap().is_none());
2640
2641 assert_eq!(vec.len(), 352);
2642 }
2643
2644 #[test]
2645 fn test_fixed_length_index() {
2646 let message_type = "
2647 message test_schema {
2648 OPTIONAL FIXED_LEN_BYTE_ARRAY (11) value (DECIMAL(25,2));
2649 }
2650 ";
2651
2652 let schema = parse_message_type(message_type).unwrap();
2653 let mut out = Vec::with_capacity(1024);
2654 let mut writer =
2655 SerializedFileWriter::new(&mut out, Arc::new(schema), Default::default()).unwrap();
2656
2657 let mut r = writer.next_row_group().unwrap();
2658 let mut c = r.next_column().unwrap().unwrap();
2659 c.typed::<FixedLenByteArrayType>()
2660 .write_batch(
2661 &[vec![0; 11].into(), vec![5; 11].into(), vec![3; 11].into()],
2662 Some(&[1, 1, 0, 1]),
2663 None,
2664 )
2665 .unwrap();
2666 c.close().unwrap();
2667 r.close().unwrap();
2668 writer.close().unwrap();
2669
2670 let b = Bytes::from(out);
2671 let options = ReadOptionsBuilder::new().with_page_index().build();
2672 let reader = SerializedFileReader::new_with_options(b, options).unwrap();
2673 let page_index = reader.metadata().page_index().unwrap();
2674
2675 match page_index.column_index(0, 0) {
2676 Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(v)) => {
2677 assert_eq!(v.num_pages(), 1);
2678 assert_eq!(v.null_count(0).unwrap(), 1);
2679 assert_eq!(v.min_value(0).unwrap(), &[0; 11]);
2680 assert_eq!(v.max_value(0).unwrap(), &[5; 11]);
2681 }
2682 _ => unreachable!(),
2683 }
2684 }
2685
2686 #[test]
2687 fn test_multi_gz() {
2688 let file = get_test_file("concatenated_gzip_members.parquet");
2689 let reader = SerializedFileReader::new(file).unwrap();
2690 let row_group_reader = reader.get_row_group(0).unwrap();
2691 match row_group_reader.get_column_reader(0).unwrap() {
2692 ColumnReader::Int64ColumnReader(mut reader) => {
2693 let mut buffer = Vec::with_capacity(1024);
2694 let mut def_levels = Vec::with_capacity(1024);
2695 let (num_records, num_values, num_levels) = reader
2696 .read_records(1024, Some(&mut def_levels), None, &mut buffer)
2697 .unwrap();
2698
2699 assert_eq!(num_records, 513);
2700 assert_eq!(num_values, 513);
2701 assert_eq!(num_levels, 513);
2702
2703 let expected: Vec<i64> = (1..514).collect();
2704 assert_eq!(&buffer, &expected);
2705 }
2706 _ => unreachable!(),
2707 }
2708 }
2709
2710 #[test]
2711 #[cfg_attr(miri, ignore)] fn test_byte_stream_split_extended() {
2713 let path = format!(
2714 "{}/byte_stream_split_extended.gzip.parquet",
2715 arrow::util::test_util::parquet_test_data(),
2716 );
2717 let file = File::open(path).unwrap();
2718 let reader = Box::new(SerializedFileReader::new(file).expect("Failed to create reader"));
2719
2720 let mut iter = reader
2722 .get_row_iter(None)
2723 .expect("Failed to create row iterator");
2724
2725 let mut start = 0;
2726 let end = reader.metadata().file_metadata().num_rows();
2727
2728 let check_row = |row: Result<Row, ParquetError>| {
2729 assert!(row.is_ok());
2730 let r = row.unwrap();
2731 assert_eq!(r.get_float16(0).unwrap(), r.get_float16(1).unwrap());
2732 assert_eq!(r.get_float(2).unwrap(), r.get_float(3).unwrap());
2733 assert_eq!(r.get_double(4).unwrap(), r.get_double(5).unwrap());
2734 assert_eq!(r.get_int(6).unwrap(), r.get_int(7).unwrap());
2735 assert_eq!(r.get_long(8).unwrap(), r.get_long(9).unwrap());
2736 assert_eq!(r.get_bytes(10).unwrap(), r.get_bytes(11).unwrap());
2737 assert_eq!(r.get_decimal(12).unwrap(), r.get_decimal(13).unwrap());
2738 };
2739
2740 while start < end {
2741 match iter.next() {
2742 Some(row) => check_row(row),
2743 None => break,
2744 }
2745 start += 1;
2746 }
2747 }
2748
2749 #[test]
2750 fn test_filtered_rowgroup_metadata() {
2751 let message_type = "
2752 message test_schema {
2753 REQUIRED INT32 a;
2754 }
2755 ";
2756 let schema = Arc::new(parse_message_type(message_type).unwrap());
2757 let props = Arc::new(
2758 WriterProperties::builder()
2759 .set_statistics_enabled(EnabledStatistics::Page)
2760 .build(),
2761 );
2762 let mut file: File = tempfile::tempfile().unwrap();
2763 let mut file_writer = SerializedFileWriter::new(&mut file, schema, props).unwrap();
2764 let data = [1, 2, 3, 4, 5];
2765
2766 for idx in 0..5 {
2768 let data_i: Vec<i32> = data.iter().map(|x| x * (idx + 1)).collect();
2769 let mut row_group_writer = file_writer.next_row_group().unwrap();
2770 if let Some(mut writer) = row_group_writer.next_column().unwrap() {
2771 writer
2772 .typed::<Int32Type>()
2773 .write_batch(data_i.as_slice(), None, None)
2774 .unwrap();
2775 writer.close().unwrap();
2776 }
2777 row_group_writer.close().unwrap();
2778 file_writer.flushed_row_groups();
2779 }
2780 let file_metadata = file_writer.close().unwrap();
2781
2782 assert_eq!(file_metadata.file_metadata().num_rows(), 25);
2783 assert_eq!(file_metadata.num_row_groups(), 5);
2784
2785 let read_options = ReadOptionsBuilder::new()
2787 .with_page_index()
2788 .with_predicate(Box::new(|rgmeta, _| rgmeta.ordinal().unwrap_or(0) == 2))
2789 .build();
2790 let reader =
2791 SerializedFileReader::new_with_options(file.try_clone().unwrap(), read_options)
2792 .unwrap();
2793 let metadata = reader.metadata();
2794
2795 assert_eq!(metadata.num_row_groups(), 1);
2797 assert_eq!(metadata.row_group(0).ordinal(), Some(2));
2798
2799 assert!(metadata.page_index().is_some_and(|idx| idx.is_complete()));
2801 let page_index = metadata.page_index().unwrap();
2802
2803 let col_stats = metadata.row_group(0).column(0).statistics().unwrap();
2804 let pg_idx = page_index.column_index(0, 0);
2805 let off_idx_i = page_index.offset_index(0, 0);
2806
2807 match pg_idx {
2809 Some(ColumnIndexMetaData::INT32(int_idx)) => {
2810 let min = col_stats.min_bytes_opt().unwrap().get_i32_le();
2811 let max = col_stats.max_bytes_opt().unwrap().get_i32_le();
2812 assert_eq!(int_idx.min_value(0), Some(min).as_ref());
2813 assert_eq!(int_idx.max_value(0), Some(max).as_ref());
2814 }
2815 _ => panic!("wrong stats type"),
2816 }
2817
2818 assert_eq!(
2820 off_idx_i.as_ref().unwrap().page_locations[0].offset,
2821 metadata.row_group(0).column(0).data_page_offset()
2822 );
2823
2824 let read_options = ReadOptionsBuilder::new()
2826 .with_page_index()
2827 .with_predicate(Box::new(|rgmeta, _| rgmeta.ordinal().unwrap_or(0) % 2 == 1))
2828 .build();
2829 let reader =
2830 SerializedFileReader::new_with_options(file.try_clone().unwrap(), read_options)
2831 .unwrap();
2832 let metadata = reader.metadata();
2833
2834 assert_eq!(metadata.num_row_groups(), 2);
2836 assert_eq!(metadata.row_group(0).ordinal(), Some(1));
2837 assert_eq!(metadata.row_group(1).ordinal(), Some(3));
2838
2839 assert!(metadata.page_index().is_some_and(|idx| idx.is_complete()));
2841
2842 let page_index = metadata.page_index().unwrap();
2843
2844 for rg_idx in 0..metadata.num_row_groups() {
2845 let col_stats = metadata.row_group(rg_idx).column(0).statistics().unwrap();
2846 let pg_idx = page_index.column_index(rg_idx, 0);
2847 let off_idx_i = page_index.offset_index(rg_idx, 0);
2848
2849 match pg_idx {
2851 Some(ColumnIndexMetaData::INT32(int_idx)) => {
2852 let min = col_stats.min_bytes_opt().unwrap().get_i32_le();
2853 let max = col_stats.max_bytes_opt().unwrap().get_i32_le();
2854 assert_eq!(int_idx.min_value(0), Some(min).as_ref());
2855 assert_eq!(int_idx.max_value(0), Some(max).as_ref());
2856 }
2857 _ => panic!("wrong stats type"),
2858 }
2859
2860 assert_eq!(
2862 off_idx_i.as_ref().unwrap().page_locations[0].offset,
2863 metadata.row_group(rg_idx).column(0).data_page_offset()
2864 );
2865 }
2866 }
2867
2868 #[test]
2869 fn test_reuse_schema() {
2870 let file = get_test_file("alltypes_plain.parquet");
2871 let file_reader = SerializedFileReader::new(file.try_clone().unwrap()).unwrap();
2872 let schema = file_reader.metadata().file_metadata().schema_descr_ptr();
2873 let expected = file_reader.metadata;
2874
2875 let options = ReadOptionsBuilder::new()
2876 .with_parquet_schema(schema)
2877 .build();
2878 let file_reader = SerializedFileReader::new_with_options(file, options).unwrap();
2879
2880 assert_eq!(expected.as_ref(), file_reader.metadata.as_ref());
2881 assert!(Arc::ptr_eq(
2883 &expected.file_metadata().schema_descr_ptr(),
2884 &file_reader.metadata.file_metadata().schema_descr_ptr()
2885 ));
2886 }
2887
2888 #[test]
2889 fn test_read_unknown_logical_type() {
2890 let file = get_test_file("unknown-logical-type.parquet");
2891 let reader = SerializedFileReader::new(file).expect("Error opening file");
2892
2893 let schema = reader.metadata().file_metadata().schema_descr();
2894 assert_eq!(
2895 schema.column(0).logical_type_ref(),
2896 Some(&basic::LogicalType::String)
2897 );
2898 assert_eq!(
2899 schema.column(1).logical_type_ref(),
2900 Some(&basic::LogicalType::_Unknown { field_id: 2555 })
2901 );
2902 assert_eq!(schema.column(1).physical_type(), Type::BYTE_ARRAY);
2903
2904 let mut iter = reader
2905 .get_row_iter(None)
2906 .expect("Failed to create row iterator");
2907
2908 let mut num_rows = 0;
2909 while iter.next().is_some() {
2910 num_rows += 1;
2911 }
2912 assert_eq!(num_rows, reader.metadata().file_metadata().num_rows());
2913 }
2914}