fix(cube): Use reader_options_customizer in one more place

Author: srh

datafusion/core/src/datasource/file_format/parquet.rs

@@ -31,7 +31,7 @@ use crate::arrow::array::RecordBatch;
 use crate::arrow::datatypes::{Fields, Schema, SchemaRef};
 use crate::datasource::file_format::file_compression_type::FileCompressionType;
 use crate::datasource::file_format::write::get_writer_schema;
-use crate::datasource::physical_plan::parquet::{can_expr_be_pushed_down_with_schemas, get_reader_options_config_or_default, MetadataFetcher, ReaderOptionsConfig};
+use crate::datasource::physical_plan::parquet::{can_expr_be_pushed_down_with_schemas, get_reader_options_config_or_default, get_reader_options_customizer, MetadataFetcher};
 use crate::datasource::physical_plan::parquet::source::ParquetSource;
 use crate::datasource::physical_plan::{FileSink, FileSinkConfig};
 use crate::datasource::statistics::{create_max_min_accs, get_col_stats};
@@ -432,7 +432,7 @@ impl FileFormat for ParquetFormat {
 
     async fn create_physical_plan(
         &self,
-        _state: &dyn Session,
+        state: &dyn Session,
         conf: FileScanConfig,
         filters: Option<&Arc<dyn PhysicalExpr>>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
@@ -451,7 +451,7 @@ impl FileFormat for ParquetFormat {
             metadata_size_hint = Some(metadata);
         }
 
-        let mut source = ParquetSource::new(self.options.clone(), ReaderOptionsConfig::noop());  // TODO upgrade DF
+        let mut source = ParquetSource::new(self.options.clone(), get_reader_options_customizer(state.config()));
 
         if let Some(predicate) = predicate {
             source = source.with_predicate(Arc::clone(&conf.file_schema), predicate);

datafusion/core/src/datasource/physical_plan/parquet/mod.rs

@@ -523,18 +523,9 @@ impl ExecutionPlan for ParquetExec {
         partition_index: usize,
         ctx: Arc<TaskContext>,
     ) -> Result<SendableRecordBatchStream> {
-        // TODO upgrade DF: we need  reader_options_customizer used as in the commented code below.
         self.inner.execute(partition_index, ctx)
 
         // TODO upgrade DF:  We also want Ok(Box::pin(stream.instrument(tracing::trace_span!("read_files")))) applied, probably in inner.
-
-        // let reader_options_customizer =
-        //     get_reader_options_customizer(ctx.session_config());
-
-        // let opener = ParquetOpener {
-        // ...
-        //     reader_options_customizer,
-        // };
     }
     fn metrics(&self) -> Option<MetricsSet> {
         self.inner.metrics()

datafusion/core/src/datasource/physical_plan/parquet/source.rs

@@ -300,8 +300,16 @@ impl ParquetSource {
     pub fn new(table_parquet_options: TableParquetOptions, reader_options_customizer: Arc<dyn ReaderOptionsCustomizer>) -> Self {
         Self {
             table_parquet_options,
+            metrics: Default::default(),
+            predicate: Default::default(),
+            pruning_predicate: Default::default(),
+            page_pruning_predicate: Default::default(),
+            parquet_file_reader_factory: Default::default(),
+            schema_adapter_factory: Default::default(),
+            batch_size: Default::default(),
+            metadata_size_hint: Default::default(),
+            projected_statistics: Default::default(),
             reader_options_customizer,
-            ..Self::default()
         }
     }