create PageIndexPolicy to allow optional indexes (#8071)

# Which issue does this PR close?

- Closes #8070.

# Rationale for this change

This change introduces a more flexible way to handle page indexes
(column and offset indexes) in Parquet files. Previously, the reading of
these indexes was controlled by boolean flags, which indicated read
required or do not read. The new `PageIndexPolicy` enum (`Off`,
`Optional`, `Required`) provides finer control, allowing users to
specify whether an index is not read, read if present (without error if
missing), or strictly required (error if missing).

# What changes are included in this PR?

- Introduced a new `PageIndexPolicy` enum with `Off`, `Optional`, and
`Required` variants.
- Replaced the boolean `column_index` and `offset_index` fields in
`ParquetMetaDataReader` with the new `PageIndexPolicy` enum.
- Updated the `ParquetMetaDataReader::new()` function to initialize page
index policies to `Off`, preserving previous defaults.
- Modified existing `with_page_indexes`, `with_column_indexes`, and
`with_offset_indexes` methods to utilize the new `PageIndexPolicy`,
defaulting to `Required` when enabling indexes.
- Added new methods: `with_page_index_policy`,
`with_column_index_policy`, and `with_offset_index_policy` to allow
direct setting of the page index policy.
- Adjusted the internal logic for parsing column and offset indexes to
respect the specified `PageIndexPolicy`, including returning an error if
a `Required` index is not found.

# Are these changes tested?

Yes, a new test file `parquet/tests/page_index.rs` has been added to
cover the functionality of the new `PageIndexPolicy` and its integration
with `ParquetMetaDataReader`.

# Are there any user-facing changes?

Yes, there are user-facing changes to the `ParquetMetaDataReader` API.
The `with_column_indexes` and `with_offset_indexes` methods now
implicitly use `PageIndexPolicy::Required` when enabling page indexes.
New methods `with_page_index_policy`, `with_column_index_policy`, and
`with_offset_index_policy` have been added.

Author: kczimm

parquet/benches/arrow_reader_row_filter.rs

@@ -70,7 +70,7 @@ use parquet::arrow::arrow_reader::{
 use parquet::arrow::async_reader::AsyncFileReader;
 use parquet::arrow::{ArrowWriter, ParquetRecordBatchStreamBuilder, ProjectionMask};
 use parquet::basic::Compression;
-use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
+use parquet::file::metadata::{PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader};
 use parquet::file::properties::WriterProperties;
 use rand::{rngs::StdRng, Rng, SeedableRng};
 use std::ops::Range;
@@ -550,7 +550,8 @@ struct InMemoryReader {
 
 impl InMemoryReader {
     fn try_new(inner: &Bytes) -> parquet::errors::Result<Self> {
-        let mut metadata_reader = ParquetMetaDataReader::new().with_page_indexes(true);
+        let mut metadata_reader =
+            ParquetMetaDataReader::new().with_page_index_policy(PageIndexPolicy::Required);
         metadata_reader.try_parse(inner)?;
         let metadata = metadata_reader.finish().map(Arc::new)?;
 

parquet/examples/external_metadata.rs

@@ -20,7 +20,9 @@ use arrow_cast::pretty::pretty_format_batches;
 use futures::TryStreamExt;
 use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ArrowReaderOptions};
 use parquet::arrow::{ArrowWriter, ParquetRecordBatchStreamBuilder};
-use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader, ParquetMetaDataWriter};
+use parquet::file::metadata::{
+    PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader, ParquetMetaDataWriter,
+};
 use parquet::file::properties::{EnabledStatistics, WriterProperties};
 use std::fs::File;
 use std::path::{Path, PathBuf};
@@ -111,7 +113,7 @@ async fn get_metadata_from_remote_parquet_file(
 
     // tell the reader to read the page index
     ParquetMetaDataReader::new()
-        .with_page_indexes(true)
+        .with_page_index_policy(PageIndexPolicy::Required)
         .load_and_finish(remote_file, file_size)
         .await
         .unwrap()
@@ -160,7 +162,7 @@ fn write_metadata_to_local_file(metadata: ParquetMetaData, file: impl AsRef<Path
 fn read_metadata_from_local_file(file: impl AsRef<Path>) -> ParquetMetaData {
     let file = File::open(file).unwrap();
     ParquetMetaDataReader::new()
-        .with_page_indexes(true)
+        .with_page_index_policy(PageIndexPolicy::Required)
         .parse_and_finish(&file)
         .unwrap()
 }

parquet/src/arrow/arrow_reader/mod.rs

@@ -37,7 +37,7 @@ use crate::column::page::{PageIterator, PageReader};
 #[cfg(feature = "encryption")]
 use crate::encryption::decrypt::FileDecryptionProperties;
 use crate::errors::{ParquetError, Result};
-use crate::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
+use crate::file::metadata::{PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader};
 use crate::file::reader::{ChunkReader, SerializedPageReader};
 use crate::format::{BloomFilterAlgorithm, BloomFilterCompression, BloomFilterHash};
 use crate::schema::types::SchemaDescriptor;
@@ -383,8 +383,8 @@ pub struct ArrowReaderOptions {
     ///
     /// [ARROW_SCHEMA_META_KEY]: crate::arrow::ARROW_SCHEMA_META_KEY
     supplied_schema: Option<SchemaRef>,
-    /// If true, attempt to read `OffsetIndex` and `ColumnIndex`
-    pub(crate) page_index: bool,
+    /// Policy for reading offset and column indexes.
+    pub(crate) page_index_policy: PageIndexPolicy,
     /// If encryption is enabled, the file decryption properties can be provided
     #[cfg(feature = "encryption")]
     pub(crate) file_decryption_properties: Option<FileDecryptionProperties>,
@@ -486,7 +486,20 @@ impl ArrowReaderOptions {
     /// [`ParquetMetaData::column_index`]: crate::file::metadata::ParquetMetaData::column_index
     /// [`ParquetMetaData::offset_index`]: crate::file::metadata::ParquetMetaData::offset_index
     pub fn with_page_index(self, page_index: bool) -> Self {
-        Self { page_index, ..self }
+        let page_index_policy = PageIndexPolicy::from(page_index);
+
+        Self {
+            page_index_policy,
+            ..self
+        }
+    }
+
+    /// Set the [`PageIndexPolicy`] to determine how page indexes should be read.
+    pub fn with_page_index_policy(self, policy: PageIndexPolicy) -> Self {
+        Self {
+            page_index_policy: policy,
+            ..self
+        }
     }
 
     /// Provide the file decryption properties to use when reading encrypted parquet files.
@@ -507,7 +520,7 @@ impl ArrowReaderOptions {
     ///
     /// This can be set via [`with_page_index`][Self::with_page_index].
     pub fn page_index(&self) -> bool {
-        self.page_index
+        self.page_index_policy != PageIndexPolicy::Skip
     }
 
     /// Retrieve the currently set file decryption properties.
@@ -556,7 +569,8 @@ impl ArrowReaderMetadata {
     /// `Self::metadata` is missing the page index, this function will attempt
     /// to load the page index by making an object store request.
     pub fn load<T: ChunkReader>(reader: &T, options: ArrowReaderOptions) -> Result<Self> {
-        let metadata = ParquetMetaDataReader::new().with_page_indexes(options.page_index);
+        let metadata =
+            ParquetMetaDataReader::new().with_page_index_policy(options.page_index_policy);
         #[cfg(feature = "encryption")]
         let metadata =
             metadata.with_decryption_properties(options.file_decryption_properties.as_ref());

parquet/src/arrow/async_reader/mod.rs

@@ -52,7 +52,7 @@ use crate::bloom_filter::{
 };
 use crate::column::page::{PageIterator, PageReader};
 use crate::errors::{ParquetError, Result};
-use crate::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
+use crate::file::metadata::{PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader};
 use crate::file::page_index::offset_index::OffsetIndexMetaData;
 use crate::file::reader::{ChunkReader, Length, SerializedPageReader};
 use crate::format::{BloomFilterAlgorithm, BloomFilterCompression, BloomFilterHash};
@@ -175,8 +175,9 @@ impl<T: AsyncRead + AsyncSeek + Unpin + Send> AsyncFileReader for T {
         options: Option<&'a ArrowReaderOptions>,
     ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>>> {
         async move {
-            let metadata_reader = ParquetMetaDataReader::new()
-                .with_page_indexes(options.is_some_and(|o| o.page_index));
+            let metadata_reader = ParquetMetaDataReader::new().with_page_index_policy(
+                PageIndexPolicy::from(options.is_some_and(|o| o.page_index())),
+            );
 
             #[cfg(feature = "encryption")]
             let metadata_reader = metadata_reader.with_decryption_properties(
@@ -1262,8 +1263,9 @@ mod tests {
             &'a mut self,
             options: Option<&'a ArrowReaderOptions>,
         ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>>> {
-            let metadata_reader = ParquetMetaDataReader::new()
-                .with_page_indexes(options.is_some_and(|o| o.page_index));
+            let metadata_reader = ParquetMetaDataReader::new().with_page_index_policy(
+                PageIndexPolicy::from(options.is_some_and(|o| o.page_index())),
+            );
             self.metadata = Some(Arc::new(
                 metadata_reader.parse_and_finish(&self.data).unwrap(),
             ));
@@ -1931,6 +1933,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[allow(deprecated)]
     async fn test_in_memory_row_group_sparse() {
         let testdata = arrow::util::test_util::parquet_test_data();
         let path = format!("{testdata}/alltypes_tiny_pages.parquet");
@@ -2458,6 +2461,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[allow(deprecated)]
     async fn empty_offset_index_doesnt_panic_in_read_row_group() {
         use tokio::fs::File;
         let testdata = arrow::util::test_util::parquet_test_data();
@@ -2483,6 +2487,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[allow(deprecated)]
     async fn non_empty_offset_index_doesnt_panic_in_read_row_group() {
         use tokio::fs::File;
         let testdata = arrow::util::test_util::parquet_test_data();
@@ -2507,6 +2512,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[allow(deprecated)]
     async fn empty_offset_index_doesnt_panic_in_column_chunks() {
         use tempfile::TempDir;
         use tokio::fs::File;

parquet/src/arrow/async_reader/store.rs

@@ -20,7 +20,7 @@ use std::{ops::Range, sync::Arc};
 use crate::arrow::arrow_reader::ArrowReaderOptions;
 use crate::arrow::async_reader::{AsyncFileReader, MetadataSuffixFetch};
 use crate::errors::{ParquetError, Result};
-use crate::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
+use crate::file::metadata::{PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader};
 use bytes::Bytes;
 use futures::{future::BoxFuture, FutureExt, TryFutureExt};
 use object_store::{path::Path, ObjectStore};
@@ -200,8 +200,8 @@ impl AsyncFileReader for ParquetObjectReader {
     ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>>> {
         Box::pin(async move {
             let mut metadata = ParquetMetaDataReader::new()
-                .with_column_indexes(self.preload_column_index)
-                .with_offset_indexes(self.preload_offset_index)
+                .with_column_index_policy(PageIndexPolicy::from(self.preload_column_index))
+                .with_offset_index_policy(PageIndexPolicy::from(self.preload_offset_index))
                 .with_prefetch_hint(self.metadata_size_hint);
 
             #[cfg(feature = "encryption")]

parquet/src/arrow/mod.rs

@@ -467,6 +467,7 @@ mod test {
     use super::ProjectionMask;
 
     #[test]
+    #[allow(deprecated)]
     // Reproducer for https://github.com/apache/arrow-rs/issues/6464
     fn test_metadata_read_write_partial_offset() {
         let parquet_bytes = create_parquet_file();
@@ -514,6 +515,7 @@ mod test {
     }
 
     #[test]
+    #[allow(deprecated)]
     fn test_metadata_read_write_roundtrip_page_index() {
         let parquet_bytes = create_parquet_file();
 

parquet/src/file/metadata/mod.rs

@@ -120,7 +120,7 @@ use crate::schema::types::{
 };
 #[cfg(feature = "encryption")]
 use crate::thrift::{TCompactSliceInputProtocol, TSerializable};
-pub use reader::{FooterTail, ParquetMetaDataReader};
+pub use reader::{FooterTail, PageIndexPolicy, ParquetMetaDataReader};
 use std::ops::Range;
 use std::sync::Arc;
 pub use writer::ParquetMetaDataWriter;