You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/06/20 10:15:53 UTC
(datafusion) branch main updated: Skip loading Parquet page index when row-group statistics already prove it cannot prune (#22857)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 1fd29c9391 Skip loading Parquet page index when row-group statistics already prove it cannot prune (#22857)
1fd29c9391 is described below
commit 1fd29c9391023a33f4ef9b55d21e50588b6e840d
Author: Ratul Dawar <ra...@gmail.com>
AuthorDate: Sat Jun 20 15:31:05 2026 +0530
Skip loading Parquet page index when row-group statistics already prove it cannot prune (#22857)
## Which issue does this PR close?
- Closes #22795
## Rationale for this change
The Parquet opener was loading the page index (ColumnIndex +
OffsetIndex) before row-group statistics pruning. When all surviving row
groups are fully matched by row-group statistics (for example, `IS NOT
NULL` on a non-null column), page index I/O cannot prune further and is
wasted.
## What changes are included in this PR?
- Reorder the opener state machine: `PrepareFilters →
PruneWithStatistics → LoadPageIndex? → LoadBloomFilters`
- Skip `load_page_index` when there is no page-pruning predicate, no
surviving row groups, or every surviving row group is fully matched
- Add unit and integration tests for the gate and the fully-matched `IS
NOT NULL` case
## Are these changes tested?
- `cargo test -p datafusion-datasource-parquet should_load`
- `cargo test -p datafusion-datasource-parquet page_index_skip`
- `cargo test -p datafusion-datasource-parquet
opener::test::test_page_pruning`
- `cargo test -p datafusion --test parquet_integration`
- `cargo clippy -p datafusion-datasource-parquet --all-targets -- -D
warnings`
## Are there any user-facing changes?
No user-facing API changes. This reduces unnecessary Parquet page index
I/O during scan planning when row-group statistics already prove no
further pruning is possible.
Made with [Cursor](https://cursor.com)
---------
Co-authored-by: Cursor <cu...@cursor.com>
---
datafusion-cli/src/main.rs | 8 +-
datafusion/datasource-parquet/src/metadata.rs | 128 ++++++--
datafusion/datasource-parquet/src/metrics.rs | 19 ++
datafusion/datasource-parquet/src/opener/mod.rs | 343 ++++++++++++++++++---
datafusion/datasource-parquet/src/reader.rs | 11 +-
.../test_files/dynamic_filter_pushdown_config.slt | 2 +-
6 files changed, 439 insertions(+), 72 deletions(-)
diff --git a/datafusion-cli/src/main.rs b/datafusion-cli/src/main.rs
index d3c5d78040..a86896c41d 100644
--- a/datafusion-cli/src/main.rs
+++ b/datafusion-cli/src/main.rs
@@ -641,9 +641,9 @@ mod tests {
+-----------------------------------+-----------------+---------------------+------+------------------+
| filename | file_size_bytes | metadata_size_bytes | hits | extra |
+-----------------------------------+-----------------+---------------------+------+------------------+
- | alltypes_plain.parquet | 1851 | 8794 | 2 | page_index=false |
+ | alltypes_plain.parquet | 1851 | 8794 | 1 | page_index=false |
| alltypes_tiny_pages.parquet | 454233 | 268970 | 2 | page_index=true |
- | lz4_raw_compressed_larger.parquet | 380836 | 1331 | 2 | page_index=false |
+ | lz4_raw_compressed_larger.parquet | 380836 | 1331 | 1 | page_index=false |
+-----------------------------------+-----------------+---------------------+------+------------------+
");
@@ -672,9 +672,9 @@ mod tests {
+-----------------------------------+-----------------+---------------------+------+------------------+
| filename | file_size_bytes | metadata_size_bytes | hits | extra |
+-----------------------------------+-----------------+---------------------+------+------------------+
- | alltypes_plain.parquet | 1851 | 8794 | 5 | page_index=false |
+ | alltypes_plain.parquet | 1851 | 8794 | 4 | page_index=false |
| alltypes_tiny_pages.parquet | 454233 | 268970 | 2 | page_index=true |
- | lz4_raw_compressed_larger.parquet | 380836 | 1331 | 3 | page_index=false |
+ | lz4_raw_compressed_larger.parquet | 380836 | 1331 | 2 | page_index=false |
+-----------------------------------+-----------------+---------------------+------+------------------+
");
diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs
index 1618050a8d..ad1caa59b8 100644
--- a/datafusion/datasource-parquet/src/metadata.rs
+++ b/datafusion/datasource-parquet/src/metadata.rs
@@ -18,6 +18,7 @@
//! [`DFParquetMetadata`] for fetching Parquet file metadata, statistics
//! and schema information.
+use crate::file_format::ObjectStoreFetch;
use crate::{Int96Coercer, apply_file_schema_type_coercions};
use arrow::array::{Array, ArrayRef, BooleanArray};
use arrow::compute::kernels::cmp::eq;
@@ -42,8 +43,8 @@ use parquet::DecodeResult;
use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
use parquet::arrow::{parquet_column, parquet_to_arrow_schema};
use parquet::file::metadata::{
- PageIndexPolicy, ParquetMetaData, ParquetMetaDataPushDecoder, RowGroupMetaData,
- SortingColumn,
+ PageIndexPolicy, ParquetMetaData, ParquetMetaDataPushDecoder, ParquetMetaDataReader,
+ RowGroupMetaData, SortingColumn,
};
use parquet::file::statistics::Statistics as ParquetStatistics;
use parquet::schema::types::SchemaDescriptor;
@@ -69,6 +70,7 @@ pub struct DFParquetMetadata<'a> {
metadata_size_hint: Option<usize>,
decryption_properties: Option<Arc<FileDecryptionProperties>>,
file_metadata_cache: Option<Arc<FileMetadataCache>>,
+ page_index_policy: Option<PageIndexPolicy>,
/// timeunit to coerce INT96 timestamps to
pub coerce_int96: Option<TimeUnit>,
/// Optional timezone applied to INT96-coerced timestamps.
@@ -83,6 +85,7 @@ impl<'a> DFParquetMetadata<'a> {
metadata_size_hint: None,
decryption_properties: None,
file_metadata_cache: None,
+ page_index_policy: None,
coerce_int96: None,
coerce_int96_tz: None,
}
@@ -112,6 +115,15 @@ impl<'a> DFParquetMetadata<'a> {
self
}
+ /// Sets the policy for loading parquet page index structures (column and offset indexes).
+ pub fn with_page_index_policy(
+ mut self,
+ page_index_policy: Option<PageIndexPolicy>,
+ ) -> Self {
+ self.page_index_policy = page_index_policy;
+ self
+ }
+
/// Set timeunit to coerce INT96 timestamps to
pub fn with_coerce_int96(mut self, time_unit: Option<TimeUnit>) -> Self {
self.coerce_int96 = time_unit;
@@ -126,9 +138,27 @@ impl<'a> DFParquetMetadata<'a> {
/// Fetch parquet metadata from the remote object store
pub async fn fetch_metadata(&self) -> Result<Arc<ParquetMetaData>> {
- // implementation to fetch parquet metadata
+ // fetch_metadata
+ // │
+ // ├─ cache_metadata = encryption check
+ // ├─ page_index_policy = caller override OR default
+ // │
+ // ├─ CACHE HIT?
+ // │ │
+ // │ ├─ has index OR policy=Skip? → return cache
+ // │ │
+ // │ └─ else (footer only, wants index)
+ // │ → load_page_index (index bytes only)
+ // │ → cache_metadata() if allowed
+ // │ → return
+ // │
+ // └─ CACHE MISS
+ // → fetch_metadata_from_store(policy)
+ // → cache_metadata() if allowed
+ // → return
let cache_metadata =
!cfg!(feature = "parquet_encryption") || self.decryption_properties.is_none();
+ let page_index_policy = self.effective_page_index_policy(cache_metadata);
if cache_metadata
&& let Some(file_metadata_cache) = self.file_metadata_cache.as_ref()
@@ -139,9 +169,61 @@ impl<'a> DFParquetMetadata<'a> {
.as_any()
.downcast_ref::<CachedParquetMetaData>()
{
- return Ok(Arc::clone(cached_parquet.parquet_metadata()));
+ let cached_metadata = Arc::clone(cached_parquet.parquet_metadata());
+ // Reuse the cache when it already has page index, or when the caller
+ // asked to skip page index I/O (footer-only metadata is sufficient).
+ if Self::metadata_has_page_index(cached_metadata.as_ref())
+ || page_index_policy == PageIndexPolicy::Skip
+ {
+ return Ok(cached_metadata);
+ }
+ let metadata =
+ Self::load_page_index(self.store, self.object_meta, cached_metadata)
+ .await?;
+ if cache_metadata {
+ self.cache_metadata(Arc::clone(&metadata)).await?;
+ }
+ return Ok(metadata);
+ }
+
+ let metadata = self.fetch_metadata_from_store(page_index_policy).await?;
+ if cache_metadata {
+ self.cache_metadata(Arc::clone(&metadata)).await?;
}
+ Ok(metadata)
+ }
+
+ fn effective_page_index_policy(&self, cache_metadata: bool) -> PageIndexPolicy {
+ self.page_index_policy.unwrap_or_else(|| {
+ if cache_metadata && self.file_metadata_cache.is_some() {
+ PageIndexPolicy::Optional
+ } else {
+ PageIndexPolicy::Skip
+ }
+ })
+ }
+ fn metadata_has_page_index(metadata: &ParquetMetaData) -> bool {
+ metadata.column_index().is_some() && metadata.offset_index().is_some()
+ }
+
+ async fn cache_metadata(&self, metadata: Arc<ParquetMetaData>) -> Result<()> {
+ if let Some(file_metadata_cache) = &self.file_metadata_cache {
+ file_metadata_cache.put(
+ &self.object_meta.location,
+ CachedFileMetadataEntry::new(
+ self.object_meta.clone(),
+ Arc::new(CachedParquetMetaData::new(metadata)),
+ ),
+ );
+ }
+ Ok(())
+ }
+
+ async fn fetch_metadata_from_store(
+ &self,
+ page_index_policy: PageIndexPolicy,
+ ) -> Result<Arc<ParquetMetaData>> {
let file_size = self.object_meta.size;
let mut decoder = ParquetMetaDataPushDecoder::try_new(file_size)
.map_err(DataFusionError::from)?;
@@ -152,14 +234,8 @@ impl<'a> DFParquetMetadata<'a> {
.with_file_decryption_properties(Some(Arc::clone(decryption_properties)));
}
- if cache_metadata && self.file_metadata_cache.is_some() {
- // Need to retrieve the entire metadata for the caching to be effective.
- decoder = decoder.with_page_index_policy(PageIndexPolicy::Optional);
- } else {
- decoder = decoder.with_page_index_policy(PageIndexPolicy::Skip);
- }
+ decoder = decoder.with_page_index_policy(page_index_policy);
- // If we have a size hint, prefetch that many bytes from the end of the file
if let Some(hint) = self.metadata_size_hint {
let prefetch_start = file_size.saturating_sub(hint as u64);
let prefetch_range = prefetch_start..file_size;
@@ -198,19 +274,27 @@ impl<'a> DFParquetMetadata<'a> {
}
};
- let metadata = Arc::new(metadata);
+ Ok(Arc::new(metadata))
+ }
- if cache_metadata && let Some(file_metadata_cache) = &self.file_metadata_cache {
- file_metadata_cache.put(
- &self.object_meta.location,
- CachedFileMetadataEntry::new(
- self.object_meta.clone(),
- Arc::new(CachedParquetMetaData::new(Arc::clone(&metadata))),
- ),
- );
+ async fn load_page_index(
+ store: &dyn ObjectStore,
+ object_meta: &ObjectMeta,
+ metadata: Arc<ParquetMetaData>,
+ ) -> Result<Arc<ParquetMetaData>> {
+ if metadata.column_index().is_some() && metadata.offset_index().is_some() {
+ return Ok(metadata);
}
-
- Ok(metadata)
+ let metadata =
+ Arc::try_unwrap(metadata).unwrap_or_else(|shared| (*shared).clone());
+ let mut reader = ParquetMetaDataReader::new_with_metadata(metadata)
+ .with_page_index_policy(PageIndexPolicy::Optional);
+ let fetch = ObjectStoreFetch::new(store, object_meta);
+ reader
+ .load_page_index(fetch)
+ .await
+ .map_err(DataFusionError::from)?;
+ Ok(Arc::new(reader.finish().map_err(DataFusionError::from)?))
}
/// Read and parse the schema of the Parquet file
diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs
index 4bf009afd6..4865975ec7 100644
--- a/datafusion/datasource-parquet/src/metrics.rs
+++ b/datafusion/datasource-parquet/src/metrics.rs
@@ -243,4 +243,23 @@ impl ParquetFileMetrics {
.counter("page_index_pages_skipped_by_fully_matched", partition);
count.add(n);
}
+
+ /// Record that page index I/O was skipped because row-group statistics
+ /// already proved page index could not prune further.
+ pub(crate) fn add_page_index_load_skipped(
+ metrics: &ExecutionPlanMetricsSet,
+ partition: usize,
+ filename: &str,
+ n: usize,
+ ) {
+ if n == 0 {
+ return;
+ }
+
+ let count = MetricBuilder::new(metrics)
+ .with_new_label("filename", filename.to_string())
+ .with_type(MetricType::Summary)
+ .counter("page_index_load_skipped", partition);
+ count.add(n);
+ }
}
diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs
index 712e0ed16f..67abae69e7 100644
--- a/datafusion/datasource-parquet/src/opener/mod.rs
+++ b/datafusion/datasource-parquet/src/opener/mod.rs
@@ -334,10 +334,10 @@ impl Morselizer for ParquetMorselizer {
/// PrepareFilters
/// |
/// v
-/// LoadPageIndex
+/// PruneWithStatistics
/// |
/// v
-/// PruneWithStatistics
+/// LoadPageIndex? (skipped when all surviving row groups are fully matched)
/// |
/// v
/// LoadBloomFilters
@@ -372,10 +372,10 @@ enum ParquetOpenState {
/// Specialize any filters for the actual file schema (only known after
/// metadata is loaded)
PrepareFilters(Box<MetadataLoadedParquetOpen>),
- /// Loading [Parquet Page Index](https://parquet.apache.org/docs/file-format/pageindex/)
- LoadPageIndex(BoxFuture<'static, Result<FiltersPreparedParquetOpen>>),
/// Pruning Row Groups
PruneWithStatistics(Box<FiltersPreparedParquetOpen>),
+ /// Loading [Parquet Page Index](https://parquet.apache.org/docs/file-format/pageindex/)
+ LoadPageIndex(BoxFuture<'static, Result<RowGroupsPrunedParquetOpen>>),
/// Loading bloom filters required for row-group pruning
LoadBloomFilters(BoxFuture<'static, Result<BloomFiltersLoadedParquetOpen>>),
/// Pruning with preloaded Bloom Filters
@@ -540,19 +540,42 @@ impl ParquetOpenState {
}
ParquetOpenState::PrepareFilters(loaded) => {
let prepared_filters = loaded.prepare_filters()?;
- Ok(ParquetOpenState::LoadPageIndex(
- prepared_filters.load_page_index().boxed(),
- ))
+ Ok(ParquetOpenState::PruneWithStatistics(Box::new(
+ prepared_filters,
+ )))
+ }
+ ParquetOpenState::PruneWithStatistics(prepared) => {
+ let prepared_row_groups = (*prepared).prune_row_groups()?;
+ if should_load_page_index(
+ prepared_row_groups.prepared.page_pruning_predicate.as_ref(),
+ &prepared_row_groups.row_groups,
+ ) {
+ Ok(ParquetOpenState::LoadPageIndex(
+ prepared_row_groups.load_page_index().boxed(),
+ ))
+ } else {
+ if prepared_row_groups
+ .prepared
+ .page_pruning_predicate
+ .is_some()
+ && !prepared_row_groups.row_groups.is_empty()
+ {
+ let prepared = &prepared_row_groups.prepared.loaded.prepared;
+ ParquetFileMetrics::add_page_index_load_skipped(
+ &prepared.metrics,
+ prepared.partition_index,
+ &prepared.file_name,
+ 1,
+ );
+ }
+ Ok(ParquetOpenState::LoadBloomFilters(
+ prepared_row_groups.load_bloom_filters().boxed(),
+ ))
+ }
}
ParquetOpenState::LoadPageIndex(future) => {
Ok(ParquetOpenState::LoadPageIndex(future))
}
- ParquetOpenState::PruneWithStatistics(prepared) => {
- let prepared_row_groups = prepared.prune_row_groups()?;
- Ok(ParquetOpenState::LoadBloomFilters(
- prepared_row_groups.load_bloom_filters().boxed(),
- ))
- }
ParquetOpenState::LoadBloomFilters(future) => {
Ok(ParquetOpenState::LoadBloomFilters(future))
}
@@ -666,9 +689,9 @@ impl MorselPlanner for ParquetMorselPlanner {
}
ParquetOpenState::LoadPageIndex(future) => {
Ok(Some(Self::schedule_io(async move {
- Ok(ParquetOpenState::PruneWithStatistics(Box::new(
- future.await?,
- )))
+ Ok(ParquetOpenState::LoadBloomFilters(
+ future.await?.load_bloom_filters().boxed(),
+ ))
})))
}
ParquetOpenState::LoadBloomFilters(future) => {
@@ -1047,27 +1070,6 @@ impl MetadataLoadedParquetOpen {
}
impl FiltersPreparedParquetOpen {
- /// Load the page index if pruning requires it and metadata did not include it.
- async fn load_page_index(mut self) -> Result<Self> {
- // The page index is not stored inline in the parquet footer so the
- // metadata load above may not have read the page index structures yet.
- // If we need them for reading and they aren't yet loaded, we need to
- // load them now.
- if self.page_pruning_predicate.is_some() {
- self.loaded.reader_metadata = load_page_index(
- self.loaded.reader_metadata,
- &mut self.loaded.prepared.async_file_reader,
- self.loaded
- .options
- .clone()
- .with_page_index_policy(PageIndexPolicy::Optional),
- )
- .await?;
- }
-
- Ok(self)
- }
-
/// Prune row groups using file ranges and parquet metadata.
fn prune_row_groups(self) -> Result<RowGroupsPrunedParquetOpen> {
let loaded = &self.loaded;
@@ -1136,6 +1138,22 @@ impl FiltersPreparedParquetOpen {
}
impl RowGroupsPrunedParquetOpen {
+ /// Load the page index if pruning requires it and metadata did not include it.
+ async fn load_page_index(mut self) -> Result<Self> {
+ self.prepared.loaded.reader_metadata = load_page_index(
+ self.prepared.loaded.reader_metadata.clone(),
+ &mut self.prepared.loaded.prepared.async_file_reader,
+ self.prepared
+ .loaded
+ .options
+ .clone()
+ .with_page_index_policy(PageIndexPolicy::Optional),
+ )
+ .await?;
+
+ Ok(self)
+ }
+
/// Load bloom filters needed for pruning when enabled and a pruning predicate exists.
async fn load_bloom_filters(mut self) -> Result<BloomFiltersLoadedParquetOpen> {
let num_row_groups = self
@@ -1589,6 +1607,22 @@ pub(crate) fn build_pruning_predicates(
)
}
+/// Returns true if the page index must be loaded for page-level pruning.
+///
+/// The page index can only prune when at least one surviving row group is not
+/// fully matched by row-group statistics alone.
+fn should_load_page_index(
+ page_pruning_predicate: Option<&Arc<PagePruningAccessPlanFilter>>,
+ row_groups: &RowGroupAccessPlanFilter,
+) -> bool {
+ page_pruning_predicate.is_some_and(|_| {
+ let fully_matched = row_groups.is_fully_matched();
+ row_groups
+ .row_group_indexes()
+ .any(|idx| !fully_matched[idx])
+ })
+}
+
/// Returns a `ArrowReaderMetadata` with the page index loaded, loading
/// it from the underlying `AsyncFileReader` if necessary.
async fn load_page_index<T: AsyncFileReader>(
@@ -1624,7 +1658,10 @@ async fn load_page_index<T: AsyncFileReader>(
mod test {
use super::*;
use super::{ConstantColumns, ParquetMorselizer, constant_columns_from_stats};
- use crate::{DefaultParquetFileReaderFactory, ParquetRowSelection, RowGroupAccess};
+ use crate::{
+ CachedParquetFileReaderFactory, DefaultParquetFileReaderFactory,
+ ParquetFileReaderFactory, ParquetRowSelection, RowGroupAccess,
+ };
use arrow::array::RecordBatch;
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use bytes::{BufMut, BytesMut};
@@ -1634,6 +1671,10 @@ mod test {
};
use datafusion_datasource::morsel::{Morsel, Morselizer};
use datafusion_datasource::{PartitionedFile, TableSchema, TableSchemaBuilder};
+ use datafusion_execution::cache::cache_manager::{
+ CachedFileMetadataEntry, FileMetadataCache,
+ };
+ use datafusion_execution::cache::default_cache::DefaultCache;
use datafusion_expr::{col, lit};
use datafusion_physical_expr::{
PhysicalExpr,
@@ -1668,6 +1709,7 @@ mod test {
predicate: Option<Arc<dyn PhysicalExpr>>,
metadata_size_hint: Option<usize>,
metrics: ExecutionPlanMetricsSet,
+ parquet_file_reader_factory: Option<Arc<dyn ParquetFileReaderFactory>>,
pushdown_filters: bool,
reorder_filters: bool,
force_filter_selections: bool,
@@ -1775,6 +1817,7 @@ mod test {
predicate: None,
metadata_size_hint: None,
metrics: ExecutionPlanMetricsSet::new(),
+ parquet_file_reader_factory: None,
pushdown_filters: false,
reorder_filters: false,
force_filter_selections: false,
@@ -1858,6 +1901,19 @@ mod test {
self
}
+ fn with_metrics(mut self, metrics: ExecutionPlanMetricsSet) -> Self {
+ self.metrics = metrics;
+ self
+ }
+
+ fn with_parquet_file_reader_factory(
+ mut self,
+ factory: Arc<dyn ParquetFileReaderFactory>,
+ ) -> Self {
+ self.parquet_file_reader_factory = Some(factory);
+ self
+ }
+
/// Set a row limit.
fn with_limit(mut self, limit: usize) -> Self {
self.limit = Some(limit);
@@ -1925,9 +1981,11 @@ mod test {
table_schema,
metadata_size_hint: self.metadata_size_hint,
metrics: self.metrics,
- parquet_file_reader_factory: Arc::new(
- DefaultParquetFileReaderFactory::new(store),
- ),
+ parquet_file_reader_factory: self
+ .parquet_file_reader_factory
+ .unwrap_or_else(|| {
+ Arc::new(DefaultParquetFileReaderFactory::new(store)) as _
+ }),
pushdown_filters: self.pushdown_filters,
reorder_filters: self.reorder_filters,
force_filter_selections: self.force_filter_selections,
@@ -2136,6 +2194,18 @@ mod test {
data_len
}
+ fn counter_metric_value(metrics: &ExecutionPlanMetricsSet, name: &str) -> usize {
+ use datafusion_physical_plan::metrics::MetricValue;
+ metrics
+ .clone_inner()
+ .sum_by_name(name)
+ .map(|metric| match metric {
+ MetricValue::Count { count, .. } => count.value(),
+ _ => 0,
+ })
+ .unwrap_or(0)
+ }
+
fn make_dynamic_expr(expr: Arc<dyn PhysicalExpr>) -> Arc<dyn PhysicalExpr> {
Arc::new(DynamicFilterPhysicalExpr::new(
expr.children().into_iter().map(Arc::clone).collect(),
@@ -2914,6 +2984,197 @@ mod test {
);
}
+ #[test]
+ fn should_load_page_index_without_predicate() {
+ use crate::RowGroupAccessPlanFilter;
+ let row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(2));
+ assert!(!should_load_page_index(None, &row_groups));
+ }
+
+ #[test]
+ fn should_load_page_index_when_surviving_row_groups_not_fully_matched() {
+ use crate::RowGroupAccessPlanFilter;
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+ let predicate = logical2physical(&col("a").gt(lit(50i32)), &schema);
+ let page_predicate = build_page_pruning_predicate(&predicate, &schema);
+ let row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(2));
+ assert!(should_load_page_index(Some(&page_predicate), &row_groups));
+ }
+
+ #[test]
+ fn should_load_page_index_when_all_surviving_row_groups_fully_matched() {
+ use crate::RowGroupAccessPlanFilter;
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+ let predicate = logical2physical(&col("a").is_not_null(), &schema);
+ let page_predicate = build_page_pruning_predicate(&predicate, &schema);
+ let mut plan = ParquetAccessPlan::new_all(1);
+ plan.mark_fully_matched(0);
+ let row_groups = RowGroupAccessPlanFilter::new(plan);
+ assert!(!should_load_page_index(Some(&page_predicate), &row_groups));
+ }
+
+ #[tokio::test]
+ async fn test_page_index_skipped_when_row_groups_fully_matched() {
+ use parquet::file::properties::WriterProperties;
+
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let values: Vec<i32> = (1..=100).collect();
+ let batch = record_batch!((
+ "a",
+ Int32,
+ values.iter().map(|v| Some(*v)).collect::<Vec<_>>()
+ ))
+ .unwrap();
+ let props = WriterProperties::builder()
+ .set_data_page_row_count_limit(10)
+ .set_write_batch_size(10)
+ .build();
+ let schema = batch.schema();
+ let data_len = write_parquet_batches(
+ Arc::clone(&store),
+ "test.parquet",
+ vec![batch],
+ Some(props),
+ )
+ .await;
+ let file = PartitionedFile::new(
+ "test.parquet".to_string(),
+ u64::try_from(data_len).unwrap(),
+ );
+ let predicate = logical2physical(&col("a").gt(lit(0i32)), &schema);
+ let metrics = ExecutionPlanMetricsSet::new();
+
+ let morselizer = ParquetMorselizerBuilder::new()
+ .with_store(Arc::clone(&store))
+ .with_schema(Arc::clone(&schema))
+ .with_predicate(Arc::clone(&predicate))
+ .with_enable_page_index(true)
+ .with_row_group_stats_pruning(true)
+ .with_pushdown_filters(false)
+ .with_metrics(metrics.clone())
+ .build();
+
+ let (_, rows) =
+ count_batches_and_rows(open_file(&morselizer, file).await.unwrap()).await;
+ assert_eq!(rows, 100);
+ assert_eq!(counter_metric_value(&metrics, "page_index_load_skipped"), 1);
+ }
+
+ #[tokio::test]
+ async fn test_page_index_skipped_with_cached_reader_factory() {
+ use parquet::file::properties::WriterProperties;
+
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let metadata_cache: Arc<FileMetadataCache> =
+ Arc::new(DefaultCache::<Path, CachedFileMetadataEntry>::new(
+ 64 * 1024 * 1024,
+ ));
+ let values: Vec<i32> = (1..=100).collect();
+ let batch = record_batch!((
+ "a",
+ Int32,
+ values.iter().map(|v| Some(*v)).collect::<Vec<_>>()
+ ))
+ .unwrap();
+ let props = WriterProperties::builder()
+ .set_data_page_row_count_limit(10)
+ .set_write_batch_size(10)
+ .build();
+ let schema = batch.schema();
+ let data_len = write_parquet_batches(
+ Arc::clone(&store),
+ "test.parquet",
+ vec![batch],
+ Some(props),
+ )
+ .await;
+ let file = PartitionedFile::new(
+ "test.parquet".to_string(),
+ u64::try_from(data_len).unwrap(),
+ );
+ let predicate = logical2physical(&col("a").gt(lit(0i32)), &schema);
+ let metrics = ExecutionPlanMetricsSet::new();
+
+ let morselizer = ParquetMorselizerBuilder::new()
+ .with_store(Arc::clone(&store))
+ .with_schema(Arc::clone(&schema))
+ .with_predicate(Arc::clone(&predicate))
+ .with_enable_page_index(true)
+ .with_row_group_stats_pruning(true)
+ .with_pushdown_filters(false)
+ .with_metrics(metrics.clone())
+ .with_parquet_file_reader_factory(Arc::new(
+ CachedParquetFileReaderFactory::new(
+ Arc::clone(&store),
+ Arc::clone(&metadata_cache),
+ ),
+ ))
+ .build();
+
+ let (_, rows) =
+ count_batches_and_rows(open_file(&morselizer, file).await.unwrap()).await;
+ assert_eq!(rows, 100);
+ assert_eq!(counter_metric_value(&metrics, "page_index_load_skipped"), 1);
+
+ let cached = metadata_cache
+ .get(&Path::from("test.parquet"))
+ .expect("metadata cache should contain the file");
+ let extra_info = cached.file_metadata.extra_info();
+ let page_index_cached = extra_info.get("page_index").map(String::as_str);
+ assert_eq!(
+ page_index_cached,
+ Some("false"),
+ "cached metadata should not include page index when opener skips it"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_page_index_loaded_when_not_fully_matched() {
+ use parquet::file::properties::WriterProperties;
+
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let values: Vec<i32> = (1..=100).collect();
+ let batch = record_batch!((
+ "a",
+ Int32,
+ values.iter().map(|v| Some(*v)).collect::<Vec<_>>()
+ ))
+ .unwrap();
+ let props = WriterProperties::builder()
+ .set_data_page_row_count_limit(10)
+ .set_write_batch_size(10)
+ .build();
+ let schema = batch.schema();
+ let data_len = write_parquet_batches(
+ Arc::clone(&store),
+ "test.parquet",
+ vec![batch],
+ Some(props),
+ )
+ .await;
+ let file = PartitionedFile::new(
+ "test.parquet".to_string(),
+ u64::try_from(data_len).unwrap(),
+ );
+ let predicate = logical2physical(&col("a").gt(lit(90i32)), &schema);
+ let metrics = ExecutionPlanMetricsSet::new();
+
+ let morselizer = ParquetMorselizerBuilder::new()
+ .with_store(Arc::clone(&store))
+ .with_schema(Arc::clone(&schema))
+ .with_predicate(Arc::clone(&predicate))
+ .with_enable_page_index(true)
+ .with_pushdown_filters(false)
+ .with_row_group_stats_pruning(false)
+ .with_metrics(metrics.clone())
+ .build();
+
+ let (_, rows) =
+ count_batches_and_rows(open_file(&morselizer, file).await.unwrap()).await;
+ assert_eq!(rows, 10);
+ assert_eq!(counter_metric_value(&metrics, "page_index_load_skipped"), 0);
+ }
+
async fn fully_matched_split_test_file(
store: Arc<dyn ObjectStore>,
) -> (SchemaRef, PartitionedFile) {
diff --git a/datafusion/datasource-parquet/src/reader.rs b/datafusion/datasource-parquet/src/reader.rs
index f1d7c82b26..4df636b894 100644
--- a/datafusion/datasource-parquet/src/reader.rs
+++ b/datafusion/datasource-parquet/src/reader.rs
@@ -176,9 +176,10 @@ impl ParquetFileReaderFactory for DefaultParquetFileReaderFactory {
/// Implementation of [`ParquetFileReaderFactory`] supporting the caching of footer and page
/// metadata. Reads and updates the [`FileMetadataCache`] with the [`ParquetMetaData`] data.
-/// This reader always loads the entire metadata (including page index, unless the file is
-/// encrypted), even if not required by the current query, to ensure it is always available for
-/// those that need it.
+///
+/// [`CachedParquetFileReader::get_metadata`] forwards the [`parquet::file::metadata::PageIndexPolicy`] from
+/// [`ArrowReaderOptions`] to [`DFParquetMetadata::fetch_metadata`], so callers such as the
+/// parquet opener can skip page-index I/O during the initial metadata load.
#[derive(Debug)]
pub struct CachedParquetFileReaderFactory {
store: Arc<dyn ObjectStore>,
@@ -289,7 +290,6 @@ impl AsyncFileReader for CachedParquetFileReader {
fn get_metadata<'a>(
&'a mut self,
- #[cfg_attr(not(feature = "parquet_encryption"), expect(unused_variables))]
options: Option<&'a ArrowReaderOptions>,
) -> BoxFuture<'a, parquet::errors::Result<Arc<ParquetMetaData>>> {
let object_meta = self.partitioned_file.object_meta.clone();
@@ -304,10 +304,13 @@ impl AsyncFileReader for CachedParquetFileReader {
#[cfg(not(feature = "parquet_encryption"))]
let file_decryption_properties = None;
+ let page_index_policy = options.map(|o| o.column_index_policy());
+
DFParquetMetadata::new(&self.store, &object_meta)
.with_decryption_properties(file_decryption_properties)
.with_file_metadata_cache(Some(Arc::clone(&metadata_cache)))
.with_metadata_size_hint(self.metadata_size_hint)
+ .with_page_index_policy(page_index_policy)
.fetch_metadata()
.await
.map_err(|e| {
diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
index e436ca7952..7ddebc235a 100644
--- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
+++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
@@ -104,7 +104,7 @@ Plan with Metrics
03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, <slt:ignore>]
04)------FilterExec: value@1 > 3, metrics=[output_rows=10, <slt:ignore>, selectivity=100% (10/10)]
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, <slt:ignore>]
-06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=1 [...]
+06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=1 [...]
statement ok
set datafusion.explain.analyze_level = dev;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org