Conditionally build page pruning predicates

Page pruning predicates in the Parquet opener are constructed regardless
of whether enable_page_index is set. Under high query load, this uses significant
CPU time although these predicates are created and discarded quickly.

This commit reorders the predicate creation flow to only construct page pruning
predicates if enable_page_index is enabled. Regular predicates are created always
as before.

Author: fpetkovski

datafusion/datasource-parquet/src/opener.rs

@@ -285,7 +285,7 @@ impl FileOpener for ParquetOpener {
             }
 
             // Build predicates for this specific file
-            let (pruning_predicate, page_pruning_predicate) = build_pruning_predicates(
+            let pruning_predicate = build_pruning_predicates(
                 predicate.as_ref(),
                 &predicate_file_schema,
                 &predicate_creation_errors,
@@ -294,15 +294,26 @@ impl FileOpener for ParquetOpener {
             // The page index is not stored inline in the parquet footer so the
             // code 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 should_enable_page_index(enable_page_index, &page_pruning_predicate) {
-                reader_metadata = load_page_index(
-                    reader_metadata,
-                    &mut async_file_reader,
-                    // Since we're manually loading the page index the option here should not matter but we pass it in for consistency
-                    options.with_page_index(true),
-                )
-                .await?;
-            }
+            let page_pruning_predicate = match (enable_page_index, &predicate) {
+                (true, Some(predicate)) => {
+                    let page_pruning_predicate =
+                        build_page_pruning_predicate(&predicate, &predicate_file_schema);
+                    if page_pruning_predicate.filter_number() > 0 {
+                        reader_metadata = load_page_index(
+                            reader_metadata,
+                            &mut async_file_reader,
+                            // Since we're manually loading the page index the option here should not matter but we pass it in for consistency
+                            options.with_page_index(true),
+                        )
+                        .await?;
+
+                        Some(page_pruning_predicate)
+                    } else {
+                        None
+                    }
+                }
+                _ => None,
+            };
 
             metadata_timer.stop();
 
@@ -666,20 +677,16 @@ pub(crate) fn build_pruning_predicates(
     predicate: Option<&Arc<dyn PhysicalExpr>>,
     file_schema: &SchemaRef,
     predicate_creation_errors: &Count,
-) -> (
-    Option<Arc<PruningPredicate>>,
-    Option<Arc<PagePruningAccessPlanFilter>>,
-) {
+) -> Option<Arc<PruningPredicate>> {
     let Some(predicate) = predicate.as_ref() else {
-        return (None, None);
+        return None;
     };
     let pruning_predicate = build_pruning_predicate(
         Arc::clone(predicate),
         file_schema,
         predicate_creation_errors,
     );
-    let page_pruning_predicate = build_page_pruning_predicate(predicate, file_schema);
-    (pruning_predicate, Some(page_pruning_predicate))
+    pruning_predicate
 }
 
 /// Returns a `ArrowReaderMetadata` with the page index loaded, loading
@@ -713,18 +720,6 @@ async fn load_page_index<T: AsyncFileReader>(
     }
 }
 
-fn should_enable_page_index(
-    enable_page_index: bool,
-    page_pruning_predicate: &Option<Arc<PagePruningAccessPlanFilter>>,
-) -> bool {
-    enable_page_index
-        && page_pruning_predicate.is_some()
-        && page_pruning_predicate
-            .as_ref()
-            .map(|p| p.filter_number() > 0)
-            .unwrap_or(false)
-}
-
 #[cfg(test)]
 mod test {
     use std::sync::Arc;

datafusion/datasource-parquet/src/source.rs

@@ -659,19 +659,19 @@ impl FileSource for ParquetSource {
 
                 write!(f, "{predicate_string}")?;
 
-                // Try to build a the pruning predicates.
+                // Try to build the pruning predicates.
                 // These are only generated here because it's useful to have *some*
                 // idea of what pushdown is happening when viewing plans.
-                // However it is important to note that these predicates are *not*
+                // However, it is important to note that these predicates are *not*
                 // necessarily the predicates that are actually evaluated:
                 // the actual predicates are built in reference to the physical schema of
                 // each file, which we do not have at this point and hence cannot use.
-                // Instead we use the logical schema of the file (the table schema without partition columns).
+                // Instead, we use the logical schema of the file (the table schema without partition columns).
                 if let (Some(file_schema), Some(predicate)) =
                     (&self.file_schema, &self.predicate)
                 {
                     let predicate_creation_errors = Count::new();
-                    if let (Some(pruning_predicate), _) = build_pruning_predicates(
+                    if let Some(pruning_predicate) = build_pruning_predicates(
                         Some(predicate),
                         file_schema,
                         &predicate_creation_errors,