Skip to main content

parquet/file/
serialized_reader.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Contains implementations of the reader traits FileReader, RowGroupReader and PageReader
19//! Also contains implementations of the ChunkReader for files (with buffering) and byte arrays (RAM)
20
21use 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
80/// Conversion into a [`RowIter`]
81/// using the full file schema over all row groups.
82impl 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
91// ----------------------------------------------------------------------
92// Implementations of file & row group readers
93
94/// A serialized implementation for Parquet [`FileReader`].
95pub struct SerializedFileReader<R: ChunkReader> {
96    chunk_reader: Arc<R>,
97    metadata: Arc<ParquetMetaData>,
98    props: ReaderPropertiesPtr,
99}
100
101/// A predicate for filtering row groups, invoked with the metadata and index
102/// of each row group in the file. Only row groups for which the predicate
103/// evaluates to `true` will be scanned
104pub type ReadGroupPredicate = Box<dyn FnMut(&RowGroupMetaData, usize) -> bool>;
105
106/// A builder for [`ReadOptions`].
107/// For the predicates that are added to the builder,
108/// they will be chained using 'AND' to filter the row groups.
109#[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    /// New builder
119    pub fn new() -> Self {
120        Self::default()
121    }
122
123    /// Add a predicate on row group metadata to the reading option,
124    /// Filter only row groups that match the predicate criteria
125    pub fn with_predicate(mut self, predicate: ReadGroupPredicate) -> Self {
126        self.predicates.push(predicate);
127        self
128    }
129
130    /// Add a range predicate on filtering row groups if their midpoints are within
131    /// the Closed-Open range `[start..end) {x | start <= x < end}`
132    ///
133    /// # Panics
134    ///
135    /// Panics if `end <= start`
136    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    /// Enable reading the page index structures described in
147    /// "[Column Index] Layout to Support Page Skipping"
148    ///
149    /// [Column Index]: https://github.com/apache/parquet-format/blob/master/PageIndex.md
150    pub fn with_page_index(mut self) -> Self {
151        self.enable_page_index = true;
152        self
153    }
154
155    /// Set the [`ReaderProperties`] configuration.
156    pub fn with_reader_properties(mut self, properties: ReaderProperties) -> Self {
157        self.props = Some(properties);
158        self
159    }
160
161    /// Provide a Parquet schema to use when decoding the metadata. The schema in the Parquet
162    /// footer will be skipped.
163    pub fn with_parquet_schema(mut self, schema: SchemaDescPtr) -> Self {
164        self.metadata_options.set_schema(schema);
165        self
166    }
167
168    /// Set whether to convert the [`encoding_stats`] in the Parquet `ColumnMetaData` to a bitmask
169    /// (defaults to `false`).
170    ///
171    /// See [`ColumnChunkMetaData::page_encoding_stats_mask`] for an explanation of why this
172    /// might be desirable.
173    ///
174    /// [`encoding_stats`]:
175    /// https://github.com/apache/parquet-format/blob/786142e26740487930ddc3ec5e39d780bd930907/src/main/thrift/parquet.thrift#L917
176    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    /// Sets the decoding policy for [`encoding_stats`] in the Parquet `ColumnMetaData`.
182    ///
183    /// [`encoding_stats`]:
184    /// https://github.com/apache/parquet-format/blob/786142e26740487930ddc3ec5e39d780bd930907/src/main/thrift/parquet.thrift#L917
185    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    /// Sets the decoding policy for [`statistics`] in the Parquet `ColumnMetaData`.
191    ///
192    /// [`statistics`]:
193    /// https://github.com/apache/parquet-format/blob/786142e26740487930ddc3ec5e39d780bd930907/src/main/thrift/parquet.thrift#L912
194    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    /// Sets the decoding policy for [`size_statistics`] in the Parquet `ColumnMetaData`.
200    ///
201    /// [`size_statistics`]:
202    /// https://github.com/apache/parquet-format/blob/786142e26740487930ddc3ec5e39d780bd930907/src/main/thrift/parquet.thrift#L936
203    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    /// Seal the builder and return the read options
209    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
222/// A collection of options for reading a Parquet file.
223///
224/// Predicates are currently only supported on row group metadata.
225/// All predicates will be chained using 'AND' to filter the row groups.
226pub 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    /// Creates file reader from a Parquet file.
235    /// Returns an error if the Parquet file does not exist or is corrupt.
236    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    /// Creates file reader from a Parquet file with read options.
247    /// Returns an error if the Parquet file does not exist or is corrupt.
248    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        // Filter row groups based on the predicates
256        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 page indexes are desired, build them with the filtered set of row groups
272        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
287/// Get midpoint offset for a row group
288fn 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        // Row groups should be processed sequentially.
311        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
327/// A serialized implementation for Parquet [`RowGroupReader`].
328pub 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    /// Creates new row group reader from a file, row group metadata and custom config.
338    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    // TODO: fix PARQUET-816
373    fn get_column_page_reader(&self, i: usize) -> Result<Box<dyn PageReader>> {
374        let col = self.metadata.column(i);
375
376        // TODO(ets): push page index into page reader so we don't have to clone here
377        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    /// get bloom filter for the `i`th column
390    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
399/// Decodes a [`Page`] from the provided `buffer`
400pub(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    // Verify the 32-bit CRC checksum of the page
407    #[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    // When processing data page v2, depending on enabled compression for the
416    // page, we should account for uncompressed data ('offset') of
417    // repetition and definition levels.
418    //
419    // We always use 0 offset for other pages other than v2, `true` flag means
420    // that compression will be applied if decompressor is defined
421    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                // When is_compressed flag is missing the page is considered compressed
444                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            // decompressed size of zero corresponds to a page with no non-null values
460            // see https://github.com/apache/parquet-format/blob/master/README.md#data-pages
461            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            // For unknown page type (e.g., INDEX_PAGE), skip and read next.
524            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        /// The current byte offset in the reader
537        /// Note that offset is u64 (i.e., not usize) to support 32-bit architectures such as WASM
538        offset: u64,
539
540        /// The length of the chunk in bytes
541        /// Note that remaining_bytes is u64 (i.e., not usize) to support 32-bit architectures such as WASM
542        remaining_bytes: u64,
543
544        // If the next page header has already been "peeked", we will cache it and it`s length here
545        next_page_header: Option<Box<PageHeader>>,
546
547        /// The index of the data page within this column chunk
548        page_index: usize,
549
550        /// Whether the next page is expected to be a dictionary page
551        require_dictionary: bool,
552    },
553    Pages {
554        /// Remaining page locations
555        page_locations: VecDeque<PageLocation>,
556        /// Remaining dictionary location if any
557        dictionary_page: Option<PageLocation>,
558        /// The total number of rows in this column chunk
559        total_rows: usize,
560        /// The index of the data page within this column chunk
561        page_index: usize,
562    },
563}
564
565#[derive(Default)]
566pub(crate) struct SerializedPageReaderContext {
567    /// Controls decoding of page-level statistics
568    pub(crate) read_stats: bool,
569    /// Crypto context carrying objects required for decryption
570    #[cfg(feature = "encryption")]
571    pub(crate) crypto_context: Option<Arc<CryptoContext>>,
572}
573
574/// A serialized implementation for Parquet [`PageReader`].
575pub struct SerializedPageReader<R: ChunkReader> {
576    /// The chunk reader
577    reader: Arc<R>,
578
579    /// The compression codec for this column chunk. Only set for non-PLAIN codec.
580    decompressor: Option<Box<dyn Codec>>,
581
582    /// Column chunk type.
583    physical_type: Type,
584
585    state: SerializedPageReaderState,
586
587    context: SerializedPageReaderContext,
588}
589
590impl<R: ChunkReader> SerializedPageReader<R> {
591    /// Creates a new serialized page reader from a chunk reader and metadata
592    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    /// Stub No-op implementation when encryption is disabled.
609    #[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    /// Adds any necessary crypto context to this page reader, if encryption is enabled.
621    #[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    /// Creates a new serialized page with custom options.
642    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                // If the offset of the first page doesn't match the start of the column chunk
655                // then the preceding space must contain a dictionary page.
656                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    /// Similar to `peek_next_page`, but returns the offset of the next page instead of the page metadata.
694    /// Unlike page metadata, an offset can uniquely identify a page.
695    ///
696    /// This is used when we need to read parquet with row-filter, and we don't want to decompress the page twice.
697    /// This function allows us to check if the next page is being cached or read previously.
698    #[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                            // For unknown page type (e.g., INDEX_PAGE), skip and read next.
717                            *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                            // For unknown page type (e.g., INDEX_PAGE), skip and read next.
734                            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        /// A wrapper around a [`std::io::Read`] that keeps track of the bytes read
764        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
786/// Reads (and decrypts, if `context` carries a crypto context) the page header stored
787/// at the front of `buffer`, returning the header and the number of bytes of `buffer`
788/// it occupies.
789///
790/// This is exposed for callers that need to decode a single page directly from an
791/// already-fetched byte range, outside of the normal [`SerializedPageReader`] iteration
792/// -- e.g. decoding a dictionary page standalone.
793pub(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    // The page's compressed size should not exceed the remaining bytes that are
928    // available to read. The page's uncompressed size is the expected size
929    // after decompression, which can never be negative.
930    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                            // For unknown page type (e.g., INDEX_PAGE), skip and read next.
1061                            *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                            // For unknown page type (e.g., INDEX_PAGE), skip and read next.
1079                            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                    // The next page header has already been peeked, so just advance the offset
1132                    *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                    // If a dictionary page exists, consume it by taking it (sets to None)
1166                    dictionary_page.take();
1167                } else {
1168                    // If no dictionary page exists, simply pop the data page from page_locations
1169                    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                // V2 data pages must start at record boundaries per the parquet
1183                // spec, so the current page ends at one.
1184                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        // Valid file path
1303        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        // Invalid file path
1321        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        // This test covers the case of maintaining the correct start position in a file
1359        // stream for each column reader after initializing and moving to the next one
1360        // (without necessarily reading the entire column).
1361        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        // Now buffer each col reader, we do not expect any failures like:
1371        // General("underlying Thrift error: end of file")
1372        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        // Test contents in Parquet metadata
1385        let metadata = reader.metadata();
1386        assert_eq!(metadata.num_row_groups(), 1);
1387
1388        // Test contents in file metadata
1389        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        // Test contents in row group metadata
1401        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        // Check each column order
1406        for i in 0..row_group_metadata.num_columns() {
1407            assert_eq!(file_metadata.column_order(i), ColumnOrder::UNDEFINED);
1408        }
1409
1410        // Test row group reader
1411        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        // Test page readers
1424        // TODO: test for every column
1425        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        // Test contents in Parquet metadata
1477        let metadata = reader.metadata();
1478        assert_eq!(metadata.num_row_groups(), 1);
1479
1480        // Test contents in file metadata
1481        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        // Check each column order
1497        for i in 0..row_group_metadata.num_columns() {
1498            assert_eq!(file_metadata.column_order(i), ColumnOrder::UNDEFINED);
1499        }
1500
1501        // Test row group reader
1502        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        // Test page readers
1515        // TODO: test for every column
1516        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()); // page stats are no longer read
1554                    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)] // calls native Zstd code unsupported by Miri
1565    #[test]
1566    fn test_file_reader_empty_compressed_datapage_v2() {
1567        // this file has a compressed datapage that un-compresses to 0 bytes
1568        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        // Test contents in Parquet metadata
1574        let metadata = reader.metadata();
1575        assert_eq!(metadata.num_row_groups(), 1);
1576
1577        // Test contents in file metadata
1578        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        // Check each column order
1598        for i in 0..row_group_metadata.num_columns() {
1599            assert_eq!(file_metadata.column_order(i), expected_order);
1600        }
1601
1602        // Test row group reader
1603        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        // Test page readers
1616        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()); // page stats are no longer read
1654                    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        // this file has 0 bytes compressed datapage that un-compresses to 0 bytes
1667        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        // Test contents in Parquet metadata
1673        let metadata = reader.metadata();
1674        assert_eq!(metadata.num_row_groups(), 1);
1675
1676        // Test contents in file metadata
1677        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        // Check each column order
1697        for i in 0..row_group_metadata.num_columns() {
1698            assert_eq!(file_metadata.column_order(i), expected_order);
1699        }
1700
1701        // Test row group reader
1702        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        // Test page readers
1715        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        // read first page
1837        let page = page_iterator.next();
1838        assert!(page.is_some());
1839        assert!(page.unwrap().is_ok());
1840
1841        // reach end of file
1842        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        // read first page
1850        let page = page_iterator.next();
1851        assert!(page.is_some());
1852        assert!(page.unwrap().is_ok());
1853
1854        // reach end of file
1855        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        // file with optional metadata: bloom filters, encoding stats, column index and offset index.
1884        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        // test optional bloom filter offset
1894        assert_eq!(col0_metadata.bloom_filter_offset().unwrap(), 192);
1895
1896        // test page encoding stats
1897        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        // test optional column index offset
1904        assert_eq!(col0_metadata.column_index_offset().unwrap(), 156);
1905        assert_eq!(col0_metadata.column_index_length().unwrap(), 25);
1906
1907        // test optional offset index offset
1908        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        // test page encoding stats
1923        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        // test skipping all
1940        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        // test skipping all but one column
1956        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        // test skipping all
1978        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        // test skipping all but one column
1993        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        // test initial number of row groups
2014        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        // test initial number of row groups
2036        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)] // Takes too long
2056    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        // true, true predicate
2063        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        // true, false predicate
2075        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        // false, true predicate
2087        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        // false, false predicate
2099        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    // Use java parquet-tools get below pageIndex info
2127    // !```
2128    // parquet-tools column-index ./data_index_bloom_encoding_stats.parquet
2129    // row group 0:
2130    // column index for column String:
2131    // Boundary order: ASCENDING
2132    // page-0  :
2133    // null count                 min                                  max
2134    // 0                          Hello                                today
2135    //
2136    // offset index for column String:
2137    // page-0   :
2138    // offset   compressed size       first row index
2139    // 4               152                     0
2140    ///```
2141    //
2142    fn test_page_index_reader() {
2143        let test_file = get_test_file("data_index_bloom_encoding_stats.parquet");
2144        let builder = ReadOptionsBuilder::new();
2145        //enable read page index
2146        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        // Test contents in Parquet metadata
2151        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        // only one row group
2157        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        //only one page group
2164        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        // only one row group
2172        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)] // Takes too long
2187    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        //enable read page index
2191        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        // Test contents in Parquet metadata
2196        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        // only one row group
2203        let row_group_metadata = metadata.row_group(0);
2204
2205        //col0->id: INT32 UNCOMPRESSED DO:0 FPO:4 SZ:37325/37325/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: 0, max: 7299, num_nulls: 0]
2206        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        //col1->bool_col:BOOLEAN UNCOMPRESSED DO:0 FPO:37329 SZ:3022/3022/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: false, max: true, num_nulls: 0]
2227        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        //col2->tinyint_col: INT32 UNCOMPRESSED DO:0 FPO:40351 SZ:37325/37325/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: 0, max: 9, num_nulls: 0]
2239        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        //col4->smallint_col: INT32 UNCOMPRESSED DO:0 FPO:77676 SZ:37325/37325/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: 0, max: 9, num_nulls: 0]
2256        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        //col5->smallint_col: INT32 UNCOMPRESSED DO:0 FPO:77676 SZ:37325/37325/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: 0, max: 9, num_nulls: 0]
2273        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        //col6->bigint_col: INT64 UNCOMPRESSED DO:0 FPO:152326 SZ:71598/71598/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: 0, max: 90, num_nulls: 0]
2290        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        //col7->float_col: FLOAT UNCOMPRESSED DO:0 FPO:223924 SZ:37325/37325/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: -0.0, max: 9.9, num_nulls: 0]
2307        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        //col8->double_col: DOUBLE UNCOMPRESSED DO:0 FPO:261249 SZ:71598/71598/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: -0.0, max: 90.89999999999999, num_nulls: 0]
2324        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        //col9->date_string_col: BINARY UNCOMPRESSED DO:0 FPO:332847 SZ:111948/111948/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: 01/01/09, max: 12/31/10, num_nulls: 0]
2341        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        //col10->string_col: BINARY UNCOMPRESSED DO:0 FPO:444795 SZ:45298/45298/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: 0, max: 9, num_nulls: 0]
2358        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        //col11->timestamp_col: INT96 UNCOMPRESSED DO:0 FPO:490093 SZ:111948/111948/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[num_nulls: 0, min/max not defined]
2375        // this columns lacks an index
2376        assert!(page_index.column_index(0, 10).is_none());
2377        //col12->year: INT32 UNCOMPRESSED DO:0 FPO:602041 SZ:37325/37325/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: 2009, max: 2010, num_nulls: 0]
2378        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        //col13->month: INT32 UNCOMPRESSED DO:0 FPO:639366 SZ:37325/37325/1.00 VC:7300 ENC:BIT_PACKED,RLE,PLAIN ST:[min: 1, max: 12, num_nulls: 0]
2395        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)] // Takes too long
2451    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        // enable read page index
2455        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        // use 'string_col', Boundary order: UNORDERED, total 352 data pages and 1 dictionary page.
2462        let mut column_page_reader = row_group_reader.get_column_page_reader(9).unwrap();
2463
2464        let mut vec = vec![];
2465
2466        // Step 1: Peek and ensure dictionary page is correctly identified
2467        let meta = column_page_reader.peek_next_page().unwrap().unwrap();
2468        assert!(meta.is_dict);
2469
2470        // Step 2: Call skip_next_page to skip the dictionary page
2471        column_page_reader.skip_next_page().unwrap();
2472
2473        // Step 3: Read the next data page after skipping the dictionary page
2474        let page = column_page_reader.get_next_page().unwrap().unwrap();
2475        assert!(matches!(page.page_type(), basic::PageType::DATA_PAGE));
2476
2477        // Step 4: Continue reading remaining data pages and verify correctness
2478        for _i in 0..351 {
2479            // 352 total pages, 1 dictionary page is skipped
2480            let meta = column_page_reader.peek_next_page().unwrap().unwrap();
2481            assert!(!meta.is_dict); // Verify no dictionary page here
2482            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        // Step 5: Check if all pages are read
2489        assert!(column_page_reader.peek_next_page().unwrap().is_none());
2490        assert!(column_page_reader.get_next_page().unwrap().is_none());
2491
2492        // Step 6: Verify the number of data pages read (should be 351 data pages)
2493        assert_eq!(vec.len(), 351);
2494    }
2495
2496    #[test]
2497    #[cfg_attr(miri, ignore)] // Takes too long
2498    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        //enable read page index
2502        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        //use 'int_col', Boundary order: ASCENDING, total 325 pages.
2509        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        //check read all pages.
2521        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        // use default SerializedFileReader without read offsetIndex
2532        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        //use 'int_col', Boundary order: ASCENDING, total 325 pages.
2538        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        //check read all pages.
2551        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)] // Takes too long
2559    fn test_peek_page_with_dictionary_page() {
2560        let test_file = get_test_file("alltypes_tiny_pages.parquet");
2561        let builder = ReadOptionsBuilder::new();
2562        //enable read page index
2563        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        //use 'string_col', Boundary order: UNORDERED, total 352 data ages and 1 dictionary page.
2569        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            // have checked with `parquet-tools column-index   -c string_col  ./alltypes_tiny_pages.parquet`
2581            // page meta has two scenarios(21, 20) of num_rows expect last page has 11 rows.
2582            if i != 351 {
2583                assert!((meta.num_rows == Some(21)) || (meta.num_rows == Some(20)));
2584            } else {
2585                // last page first row index is 7290, total row count is 7300
2586                // because first row start with zero, last page row count should be 10.
2587                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        //check read all pages.
2596        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        //use 'string_col', Boundary order: UNORDERED, total 352 data ages and 1 dictionary page.
2611        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            // have checked with `parquet-tools column-index   -c string_col  ./alltypes_tiny_pages.parquet`
2623            // page meta has two scenarios(21, 20) of num_rows expect last page has 11 rows.
2624            if i != 351 {
2625                assert!((meta.num_levels == Some(21)) || (meta.num_levels == Some(20)));
2626            } else {
2627                // last page first row index is 7290, total row count is 7300
2628                // because first row start with zero, last page row count should be 10.
2629                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        //check read all pages.
2638        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)] // Takes too long
2712    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        // Use full schema as projected schema
2721        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        // write 5 row groups
2767        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        // read only the 3rd row group
2786        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        // check we got the expected row group
2796        assert_eq!(metadata.num_row_groups(), 1);
2797        assert_eq!(metadata.row_group(0).ordinal(), Some(2));
2798
2799        // check we only got the relevant page indexes
2800        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        // test that we got the index matching the row group
2808        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        // check offset index matches too
2819        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        // read non-contiguous row groups
2825        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        // check we got the expected row groups
2835        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        // check we only got the relevant page indexes
2840        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            // test that we got the index matching the row group
2850            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            // check offset index matches too
2861            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        // Should have used the same schema instance
2882        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}