remove more

Author: adriangb

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

@@ -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;
@@ -123,20 +123,13 @@ pub struct ArrowOpener {
 
 impl FileOpener for ArrowOpener {
     fn open(&self, file: PartitionedFile) -> Result<FileOpenFuture> {
-        let file_meta = FileMeta {
-            object_meta: file.object_meta.clone(),
-            range: file.range.clone(),
-            extensions: file.extensions.clone(),
-            metadata_size_hint: file.metadata_size_hint,
-        };
-
         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 = file.range.clone();
             match range {
                 None => {
-                    let r = object_store.get(file_meta.location()).await?;
+                    let r = object_store.get(&file.object_meta.location).await?;
                     match r.payload {
                         #[cfg(not(target_arch = "wasm32"))]
                         GetResultPayload::File(file, _) => {
@@ -167,7 +160,7 @@ impl FileOpener for ArrowOpener {
                         ..Default::default()
                     };
                     let get_result = object_store
-                        .get_opts(file_meta.location(), get_option)
+                        .get_opts(&file.object_meta.location, get_option)
                         .await?;
                     let footer_len_buf = get_result.bytes().await?;
                     let footer_len = arrow_ipc::reader::read_footer_length(
@@ -179,7 +172,7 @@ impl FileOpener for ArrowOpener {
                         ..Default::default()
                     };
                     let get_result = object_store
-                        .get_opts(file_meta.location(), get_option)
+                        .get_opts(&file.object_meta.location, get_option)
                         .await?;
                     let footer_buf = get_result.bytes().await?;
                     let footer = arrow_ipc::root_as_footer(
@@ -207,7 +200,7 @@ impl FileOpener for ArrowOpener {
                         })
                         .collect_vec();
                     let dict_results = object_store
-                        .get_ranges(file_meta.location(), &dict_ranges)
+                        .get_ranges(&file.object_meta.location, &dict_ranges)
                         .await?;
                     for (dict_block, dict_result) in
                         footer.dictionaries().iter().flatten().zip(dict_results)
@@ -240,7 +233,7 @@ impl FileOpener for ArrowOpener {
                         .collect_vec();
 
                     let recordbatch_results = object_store
-                        .get_ranges(file_meta.location(), &recordbatch_ranges)
+                        .get_ranges(&file.object_meta.location, &recordbatch_ranges)
                         .await?;
 
                     Ok(futures::stream::iter(

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

@@ -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/datasource-avro/src/source.rs

@@ -145,9 +145,7 @@ mod private {
     use super::*;
 
     use bytes::Buf;
-    use datafusion_datasource::{
-        file_meta::FileMeta, file_stream::FileOpenFuture, PartitionedFile,
-    };
+    use datafusion_datasource::{file_stream::FileOpenFuture, PartitionedFile};
     use futures::StreamExt;
     use object_store::{GetResultPayload, ObjectStore};
 
@@ -158,17 +156,10 @@ mod private {
 
     impl FileOpener for AvroOpener {
         fn open(&self, file: PartitionedFile) -> Result<FileOpenFuture> {
-            let file_meta = FileMeta {
-                object_meta: file.object_meta.clone(),
-                range: file.range.clone(),
-                extensions: file.extensions.clone(),
-                metadata_size_hint: file.metadata_size_hint,
-            };
-
             let config = Arc::clone(&self.config);
             let object_store = Arc::clone(&self.object_store);
             Ok(Box::pin(async move {
-                let r = object_store.get(file_meta.location()).await?;
+                let r = object_store.get(&file.object_meta.location).await?;
                 match r.payload {
                     GetResultPayload::File(file, _) => {
                         let reader = config.open(file)?;

datafusion/datasource-csv/src/source.rs

@@ -26,7 +26,6 @@ use std::task::Poll;
 
 use datafusion_datasource::decoder::{deserialize_stream, DecoderDeserializer};
 use datafusion_datasource::file_compression_type::FileCompressionType;
-use datafusion_datasource::file_meta::FileMeta;
 use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener};
 use datafusion_datasource::{
     as_file_source, calculate_range, FileRange, ListingTableUrl, PartitionedFile,
@@ -338,18 +337,11 @@ impl FileOpener for CsvOpener {
     ///  A},1,2,3,4,5,6,7,8,9\n
     ///  The lines read would be: [1, 2]
     fn open(&self, file: PartitionedFile) -> Result<FileOpenFuture> {
-        let file_meta = FileMeta {
-            object_meta: file.object_meta.clone(),
-            range: file.range.clone(),
-            extensions: file.extensions.clone(),
-            metadata_size_hint: file.metadata_size_hint,
-        };
-
         // `self.config.has_header` controls whether to skip reading the 1st line header
         // If the .csv file is read in parallel and this `CsvOpener` is only reading some middle
         // partition, then don't skip first line
         let mut csv_has_header = self.config.has_header;
-        if let Some(FileRange { start, .. }) = file_meta.range {
+        if let Some(FileRange { start, .. }) = file.range {
             if start != 0 {
                 csv_has_header = false;
             }
@@ -363,7 +355,7 @@ impl FileOpener for CsvOpener {
 
         let file_compression_type = self.file_compression_type.to_owned();
 
-        if file_meta.range.is_some() {
+        if file.range.is_some() {
             assert!(
                 !file_compression_type.is_compressed(),
                 "Reading compressed .csv in parallel is not supported"
@@ -376,8 +368,7 @@ impl FileOpener for CsvOpener {
         Ok(Box::pin(async move {
             // Current partition contains bytes [start_byte, end_byte) (might contain incomplete lines at boundaries)
 
-            let calculated_range =
-                calculate_range(&file_meta, &store, terminator).await?;
+            let calculated_range = calculate_range(&file, &store, terminator).await?;
 
             let range = match calculated_range {
                 RangeCalculation::Range(None) => None,
@@ -394,19 +385,20 @@ impl FileOpener for CsvOpener {
                 ..Default::default()
             };
 
-            let result = store.get_opts(file_meta.location(), options).await?;
+            let result = store.get_opts(&file.object_meta.location, options).await?;
 
             match result.payload {
                 #[cfg(not(target_arch = "wasm32"))]
-                GetResultPayload::File(mut file, _) => {
-                    let is_whole_file_scanned = file_meta.range.is_none();
+                GetResultPayload::File(mut result_file, _) => {
+                    let is_whole_file_scanned = file.range.is_none();
                     let decoder = if is_whole_file_scanned {
                         // Don't seek if no range as breaks FIFO files
-                        file_compression_type.convert_read(file)?
+                        file_compression_type.convert_read(result_file)?
                     } else {
-                        file.seek(SeekFrom::Start(result.range.start as _))?;
+                        result_file.seek(SeekFrom::Start(result.range.start as _))?;
                         file_compression_type.convert_read(
-                            file.take((result.range.end - result.range.start) as u64),
+                            result_file
+                                .take((result.range.end - result.range.start) as u64),
                         )?
                     };
 

datafusion/datasource-json/src/source.rs

@@ -28,7 +28,6 @@ use datafusion_common::error::{DataFusionError, Result};
 use datafusion_common_runtime::JoinSet;
 use datafusion_datasource::decoder::{deserialize_stream, DecoderDeserializer};
 use datafusion_datasource::file_compression_type::FileCompressionType;
-use datafusion_datasource::file_meta::FileMeta;
 use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener};
 use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
 use datafusion_datasource::{
@@ -177,20 +176,13 @@ impl FileOpener for JsonOpener {
     /// 1. The first line of the partition is the line in which the index of the first character >= `start`.
     /// 2. The last line of the partition is the line in which the byte at position `end - 1` resides.
     fn open(&self, file: PartitionedFile) -> Result<FileOpenFuture> {
-        let file_meta = FileMeta {
-            object_meta: file.object_meta.clone(),
-            range: file.range.clone(),
-            extensions: file.extensions.clone(),
-            metadata_size_hint: file.metadata_size_hint,
-        };
-
         let store = Arc::clone(&self.object_store);
         let schema = Arc::clone(&self.projected_schema);
         let batch_size = self.batch_size;
         let file_compression_type = self.file_compression_type.to_owned();
 
         Ok(Box::pin(async move {
-            let calculated_range = calculate_range(&file_meta, &store, None).await?;
+            let calculated_range = calculate_range(&file, &store, None).await?;
 
             let range = match calculated_range {
                 RangeCalculation::Range(None) => None,
@@ -207,17 +199,18 @@ impl FileOpener for JsonOpener {
                 ..Default::default()
             };
 
-            let result = store.get_opts(file_meta.location(), options).await?;
+            let result = store.get_opts(&file.object_meta.location, options).await?;
 
             match result.payload {
                 #[cfg(not(target_arch = "wasm32"))]
-                GetResultPayload::File(mut file, _) => {
-                    let bytes = match file_meta.range {
-                        None => file_compression_type.convert_read(file)?,
+                GetResultPayload::File(mut result_file, _) => {
+                    let bytes = match file.range {
+                        None => file_compression_type.convert_read(result_file)?,
                         Some(_) => {
-                            file.seek(SeekFrom::Start(result.range.start as _))?;
+                            result_file.seek(SeekFrom::Start(result.range.start as _))?;
                             let limit = result.range.end - result.range.start;
-                            file_compression_type.convert_read(file.take(limit as u64))?
+                            file_compression_type
+                                .convert_read(result_file.take(limit as u64))?
                         }
                     };
 

datafusion/datasource/src/mod.rs

@@ -54,7 +54,6 @@ use chrono::TimeZone;
 use datafusion_common::stats::Precision;
 use datafusion_common::{exec_datafusion_err, ColumnStatistics, Result};
 use datafusion_common::{ScalarValue, Statistics};
-use file_meta::FileMeta;
 use futures::{Stream, StreamExt};
 use object_store::{path::Path, ObjectMeta};
 use object_store::{GetOptions, GetRange, ObjectStore};
@@ -258,15 +257,15 @@ pub enum RangeCalculation {
 ///
 /// Returns an `Error` if any part of the range calculation fails, such as issues in reading from the object store or invalid range boundaries.
 pub async fn calculate_range(
-    file_meta: &FileMeta,
+    file: &PartitionedFile,
     store: &Arc<dyn ObjectStore>,
     terminator: Option<u8>,
 ) -> Result<RangeCalculation> {
-    let location = file_meta.location();
-    let file_size = file_meta.object_meta.size;
+    let location = &file.object_meta.location;
+    let file_size = file.object_meta.size;
     let newline = terminator.unwrap_or(b'\n');
 
-    match file_meta.range {
+    match file.range {
         None => Ok(RangeCalculation::Range(None)),
         Some(FileRange { start, end }) => {
             let start: u64 = start.try_into().map_err(|_| {