clean up duplicate information in FileOpener trait (apache#17956)

* clean up duplicate information in FileOpener trait

* remove unused struct

* remove more

* rename paramter

* more renames

* minimize diff

* fix renames

* fix renames

* more fixes

Author: adriangb

datafusion-examples/examples/advanced_parquet_index.rs

@@ -30,7 +30,7 @@ use datafusion::common::{
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::datasource::physical_plan::parquet::ParquetAccessPlan;
 use datafusion::datasource::physical_plan::{
-    FileMeta, FileScanConfigBuilder, ParquetFileReaderFactory, ParquetSource,
+    FileScanConfigBuilder, ParquetFileReaderFactory, ParquetSource,
 };
 use datafusion::datasource::TableProvider;
 use datafusion::execution::object_store::ObjectStoreUrl;
@@ -555,15 +555,16 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory {
     fn create_reader(
         &self,
         _partition_index: usize,
-        file_meta: FileMeta,
+        partitioned_file: PartitionedFile,
         metadata_size_hint: Option<usize>,
         _metrics: &ExecutionPlanMetricsSet,
     ) -> Result<Box<dyn AsyncFileReader + Send>> {
         // for this example we ignore the partition index and metrics
         // but in a real system you would likely use them to report details on
         // the performance of the reader.
-        let filename = file_meta
-            .location()
+        let filename = partition_file
+            .object_meta
+            .location
             .parts()
             .last()
             .expect("No path in location")
@@ -572,8 +573,8 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory {
 
         let object_store = Arc::clone(&self.object_store);
         let mut inner =
-            ParquetObjectReader::new(object_store, file_meta.object_meta.location)
-                .with_file_size(file_meta.object_meta.size);
+            ParquetObjectReader::new(object_store, partition_file.object_meta.location)
+                .with_file_size(partition_file.object_meta.size);
 
         if let Some(hint) = metadata_size_hint {
             inner = inner.with_footer_size_hint(hint)

datafusion-testing

@@ -247,6 +247,7 @@ mod tests {
         Ok(())
     }
 
+    #[rstest::rstest]
     #[cfg(feature = "parquet_encryption")]
     #[tokio::test]
     async fn roundtrip_parquet_with_encryption() -> Result<()> {

datafusion/core/"/var/folders/jj/m8gr41dn2g96dzjtlk522g200000gn/T/.tmpPJDwAs/target.csv"

@@ -18,7 +18,7 @@
 use std::any::Any;
 use std::sync::Arc;
 
-use crate::datasource::physical_plan::{FileMeta, FileOpenFuture, FileOpener};
+use crate::datasource::physical_plan::{FileOpenFuture, FileOpener};
 use crate::error::Result;
 use datafusion_datasource::as_file_source;
 use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
@@ -122,18 +122,16 @@ pub struct ArrowOpener {
 }
 
 impl FileOpener for ArrowOpener {
-    fn open(
-        &self,
-        file_meta: FileMeta,
-        _file: PartitionedFile,
-    ) -> Result<FileOpenFuture> {
+    fn open(&self, partitioned_file: PartitionedFile) -> Result<FileOpenFuture> {
         let object_store = Arc::clone(&self.object_store);
         let projection = self.projection.clone();
         Ok(Box::pin(async move {
-            let range = file_meta.range.clone();
+            let range = partitioned_file.range.clone();
             match range {
                 None => {
-                    let r = object_store.get(file_meta.location()).await?;
+                    let r = object_store
+                        .get(&partitioned_file.object_meta.location)
+                        .await?;
                     match r.payload {
                         #[cfg(not(target_arch = "wasm32"))]
                         GetResultPayload::File(file, _) => {
@@ -164,7 +162,7 @@ impl FileOpener for ArrowOpener {
                         ..Default::default()
                     };
                     let get_result = object_store
-                        .get_opts(file_meta.location(), get_option)
+                        .get_opts(&partitioned_file.object_meta.location, get_option)
                         .await?;
                     let footer_len_buf = get_result.bytes().await?;
                     let footer_len = arrow_ipc::reader::read_footer_length(
@@ -176,7 +174,7 @@ impl FileOpener for ArrowOpener {
                         ..Default::default()
                     };
                     let get_result = object_store
-                        .get_opts(file_meta.location(), get_option)
+                        .get_opts(&partitioned_file.object_meta.location, get_option)
                         .await?;
                     let footer_buf = get_result.bytes().await?;
                     let footer = arrow_ipc::root_as_footer(
@@ -204,7 +202,7 @@ impl FileOpener for ArrowOpener {
                         })
                         .collect_vec();
                     let dict_results = object_store
-                        .get_ranges(file_meta.location(), &dict_ranges)
+                        .get_ranges(&partitioned_file.object_meta.location, &dict_ranges)
                         .await?;
                     for (dict_block, dict_result) in
                         footer.dictionaries().iter().flatten().zip(dict_results)
@@ -237,7 +235,10 @@ impl FileOpener for ArrowOpener {
                         .collect_vec();
 
                     let recordbatch_results = object_store
-                        .get_ranges(file_meta.location(), &recordbatch_ranges)
+                        .get_ranges(
+                            &partitioned_file.object_meta.location,
+                            &recordbatch_ranges,
+                        )
                         .await?;
 
                     Ok(futures::stream::iter(

datafusion/core/"/var/folders/jj/m8gr41dn2g96dzjtlk522g200000gn/T/.tmpua8p7n/target.csv"

@@ -43,7 +43,6 @@ pub use csv::{CsvOpener, CsvSource};
 pub use datafusion_datasource::file::FileSource;
 pub use datafusion_datasource::file_groups::FileGroup;
 pub use datafusion_datasource::file_groups::FileGroupPartitioner;
-pub use datafusion_datasource::file_meta::FileMeta;
 pub use datafusion_datasource::file_scan_config::{
     wrap_partition_type_in_dict, wrap_partition_value_in_dict, FileScanConfig,
     FileScanConfigBuilder,

datafusion/core/src/dataframe/parquet.rs

@@ -50,7 +50,6 @@ mod tests {
     use datafusion_common::test_util::{batches_to_sort_string, batches_to_string};
     use datafusion_common::{assert_contains, Result, ScalarValue};
     use datafusion_datasource::file_format::FileFormat;
-    use datafusion_datasource::file_meta::FileMeta;
     use datafusion_datasource::file_scan_config::FileScanConfigBuilder;
     use datafusion_datasource::source::DataSourceExec;
 
@@ -2207,7 +2206,7 @@ mod tests {
         fn create_reader(
             &self,
             partition_index: usize,
-            file_meta: FileMeta,
+            partitioned_file: PartitionedFile,
             metadata_size_hint: Option<usize>,
             metrics: &ExecutionPlanMetricsSet,
         ) -> Result<Box<dyn parquet::arrow::async_reader::AsyncFileReader + Send>>
@@ -2218,7 +2217,7 @@ mod tests {
                 .push(metadata_size_hint);
             self.inner.create_reader(
                 partition_index,
-                file_meta,
+                partitioned_file,
                 metadata_size_hint,
                 metrics,
             )

datafusion/core/src/datasource/physical_plan/arrow_file.rs

@@ -26,7 +26,7 @@ use arrow::record_batch::RecordBatch;
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::datasource::object_store::ObjectStoreUrl;
 use datafusion::datasource::physical_plan::{
-    FileMeta, ParquetFileMetrics, ParquetFileReaderFactory, ParquetSource,
+    ParquetFileMetrics, ParquetFileReaderFactory, ParquetSource,
 };
 use datafusion::physical_plan::collect;
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
@@ -119,11 +119,11 @@ impl ParquetFileReaderFactory for InMemoryParquetFileReaderFactory {
     fn create_reader(
         &self,
         partition_index: usize,
-        file_meta: FileMeta,
+        partitioned_file: PartitionedFile,
         metadata_size_hint: Option<usize>,
         metrics: &ExecutionPlanMetricsSet,
     ) -> Result<Box<dyn AsyncFileReader + Send>> {
-        let metadata = file_meta
+        let metadata = partitioned_file
             .extensions
             .as_ref()
             .expect("has user defined metadata");
@@ -135,13 +135,13 @@ impl ParquetFileReaderFactory for InMemoryParquetFileReaderFactory {
 
         let parquet_file_metrics = ParquetFileMetrics::new(
             partition_index,
-            file_meta.location().as_ref(),
+            partitioned_file.object_meta.location.as_ref(),
             metrics,
         );
 
         Ok(Box::new(ParquetFileReader {
             store: Arc::clone(&self.0),
-            meta: file_meta.object_meta,
+            meta: partitioned_file.object_meta,
             metrics: parquet_file_metrics,
             metadata_size_hint,
         }))

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

@@ -20,7 +20,7 @@ use arrow::{array::RecordBatch, compute::concat_batches};
 use datafusion::{datasource::object_store::ObjectStoreUrl, physical_plan::PhysicalExpr};
 use datafusion_common::{config::ConfigOptions, internal_err, Result, Statistics};
 use datafusion_datasource::{
-    file::FileSource, file_meta::FileMeta, file_scan_config::FileScanConfig,
+    file::FileSource, file_scan_config::FileScanConfig,
     file_scan_config::FileScanConfigBuilder, file_stream::FileOpenFuture,
     file_stream::FileOpener, schema_adapter::DefaultSchemaAdapterFactory,
     schema_adapter::SchemaAdapterFactory, source::DataSourceExec, PartitionedFile,
@@ -56,11 +56,7 @@ pub struct TestOpener {
 }
 
 impl FileOpener for TestOpener {
-    fn open(
-        &self,
-        _file_meta: FileMeta,
-        _file: PartitionedFile,
-    ) -> Result<FileOpenFuture> {
+    fn open(&self, _partitioned_file: PartitionedFile) -> Result<FileOpenFuture> {
         let mut batches = self.batches.clone();
         if let Some(batch_size) = self.batch_size {
             let batch = concat_batches(&batches[0].schema(), &batches)?;