Expose remaining parquet config options into ConfigOptions (try 2) (#4427)

* Expose remaining parquet config options into ConfigOptions (try 2)

* fix: doctests

* Update config doc

* Update docs/source/user-guide/configs.md

Co-authored-by: Dan Harris <1327726+thinkharderdev@users.noreply.github.com>

* Update configuration docs

Co-authored-by: Dan Harris <1327726+thinkharderdev@users.noreply.github.com>

Author: alamb

benchmarks/src/bin/tpch.rs

@@ -396,7 +396,8 @@ async fn get_table(
             }
             "parquet" => {
                 let path = format!("{}/{}", path, table);
-                let format = ParquetFormat::default().with_enable_pruning(true);
+                let format = ParquetFormat::new(ctx.config_options())
+                    .with_enable_pruning(Some(true));
 
                 (Arc::new(format), path, DEFAULT_PARQUET_EXTENSION)
             }

datafusion-examples/examples/flight_server.rs

@@ -67,7 +67,9 @@ impl FlightService for FlightServiceImpl {
     ) -> Result<Response<SchemaResult>, Status> {
         let request = request.into_inner();
 
-        let listing_options = ListingOptions::new(Arc::new(ParquetFormat::default()));
+        let config = SessionConfig::new();
+        let listing_options =
+            ListingOptions::new(Arc::new(ParquetFormat::new(config.config_options())));
         let table_path =
             ListingTableUrl::parse(&request.path[0]).map_err(to_tonic_err)?;
 

datafusion-examples/examples/parquet_sql_multiple_files.rs

@@ -32,7 +32,8 @@ async fn main() -> Result<()> {
     let testdata = datafusion::test_util::parquet_test_data();
 
     // Configure listing options
-    let file_format = ParquetFormat::default().with_enable_pruning(true);
+    let file_format =
+        ParquetFormat::new(ctx.config_options()).with_enable_pruning(Some(true));
     let listing_options = ListingOptions::new(Arc::new(file_format))
         .with_file_extension(FileType::PARQUET.get_ext());
 

datafusion/core/src/config.rs

@@ -61,6 +61,16 @@ pub const OPT_PARQUET_REORDER_FILTERS: &str =
 pub const OPT_PARQUET_ENABLE_PAGE_INDEX: &str =
     "datafusion.execution.parquet.enable_page_index";
 
+/// Configuration option "datafusion.execution.parquet.pruning"
+pub const OPT_PARQUET_ENABLE_PRUNING: &str = "datafusion.execution.parquet.pruning";
+
+/// Configuration option "datafusion.execution.parquet.skip_metadata"
+pub const OPT_PARQUET_SKIP_METADATA: &str = "datafusion.execution.parquet.skip_metadata";
+
+/// Configuration option "datafusion.execution.parquet.metadata_size_hint"
+pub const OPT_PARQUET_METADATA_SIZE_HINT: &str =
+    "datafusion.execution.parquet.metadata_size_hint";
+
 /// Configuration option "datafusion.optimizer.skip_failed_rules"
 pub const OPT_OPTIMIZER_SKIP_FAILED_RULES: &str =
     "datafusion.optimizer.skip_failed_rules";
@@ -255,6 +265,29 @@ impl BuiltInConfigs {
                  to reduce the number of rows decoded.",
                 false,
             ),
+            ConfigDefinition::new_bool(
+                OPT_PARQUET_ENABLE_PRUNING,
+                "If true, the parquet reader attempts to skip entire row groups based \
+                 on the predicate in the query and the metadata (min/max values) stored in \
+                 the parquet file.",
+                true,
+            ),
+            ConfigDefinition::new_bool(
+                OPT_PARQUET_SKIP_METADATA,
+                "If true, the parquet reader skip the optional embedded metadata that may be in \
+                the file Schema. This setting can help avoid schema conflicts when querying \
+                multiple parquet files with schemas containing compatible types but different metadata.",
+                true,
+            ),
+            ConfigDefinition::new(
+                OPT_PARQUET_METADATA_SIZE_HINT,
+                "If specified, the parquet reader will try and fetch the last `size_hint` \
+                 bytes of the parquet file optimistically. If not specified, two read are required: \
+                 One read to fetch the 8-byte parquet footer and  \
+                 another to fetch the metadata length encoded in the footer.",
+                DataType::UInt64,
+                ScalarValue::UInt64(None),
+            ),
             ConfigDefinition::new_bool(
                 OPT_OPTIMIZER_SKIP_FAILED_RULES,
                 "When set to true, the logical plan optimizer will produce warning \
@@ -424,6 +457,12 @@ impl ConfigOptions {
         get_conf_value!(self, UInt64, key, "u64")
     }
 
+    /// get a u64 configuration option as a usize
+    pub fn get_usize(&self, key: &str) -> Option<usize> {
+        let v = get_conf_value!(self, UInt64, key, "usize");
+        v.and_then(|v| v.try_into().ok())
+    }
+
     /// get a string configuration option
     pub fn get_string(&self, key: &str) -> Option<String> {
         get_conf_value!(self, Utf8, key, "string")