simplify fn signature

Author: AdamGS

datafusion-examples/examples/csv_json_opener.rs

@@ -71,7 +71,7 @@ async fn csv_opener() -> Result<()> {
         .with_batch_size(8192)
         .with_projection(&scan_config);
 
-    let opener = config.create_file_opener(Ok(object_store), &scan_config, 0)?;
+    let opener = config.create_file_opener(object_store, &scan_config, 0)?;
 
     let mut result = vec![];
     let mut stream =

datafusion/core/src/datasource/data_source.rs

@@ -38,7 +38,7 @@ pub trait FileSource: Send + Sync {
     /// Creates a `dyn FileOpener` based on given parameters
     fn create_file_opener(
         &self,
-        object_store: datafusion_common::Result<Arc<dyn ObjectStore>>,
+        object_store: Arc<dyn ObjectStore>,
         base_config: &FileScanConfig,
         partition: usize,
     ) -> datafusion_common::Result<Arc<dyn FileOpener>>;

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

@@ -211,12 +211,12 @@ pub struct ArrowSource {
 impl FileSource for ArrowSource {
     fn create_file_opener(
         &self,
-        object_store: Result<Arc<dyn ObjectStore>>,
+        object_store: Arc<dyn ObjectStore>,
         base_config: &FileScanConfig,
         _partition: usize,
     ) -> Result<Arc<dyn FileOpener>> {
         Ok(Arc::new(ArrowOpener {
-            object_store: object_store?,
+            object_store,
             projection: base_config.file_column_projection_indices(),
         }))
     }

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

@@ -194,20 +194,20 @@ impl FileSource for AvroSource {
     #[cfg(feature = "avro")]
     fn create_file_opener(
         &self,
-        object_store: Result<Arc<dyn ObjectStore>>,
+        object_store: Arc<dyn ObjectStore>,
         _base_config: &FileScanConfig,
         _partition: usize,
     ) -> Result<Arc<dyn FileOpener>> {
         Ok(Arc::new(private::AvroOpener {
             config: Arc::new(self.clone()),
-            object_store: object_store?,
+            object_store,
         }))
     }
 
     #[cfg(not(feature = "avro"))]
     fn create_file_opener(
         &self,
-        _object_store: Result<Arc<dyn ObjectStore>>,
+        _object_store: Arc<dyn ObjectStore>,
         _base_config: &FileScanConfig,
         _partition: usize,
     ) -> Result<Arc<dyn FileOpener>> {

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

@@ -564,14 +564,14 @@ impl CsvOpener {
 impl FileSource for CsvSource {
     fn create_file_opener(
         &self,
-        object_store: Result<Arc<dyn ObjectStore>>,
+        object_store: Arc<dyn ObjectStore>,
         base_config: &FileScanConfig,
         _partition: usize,
     ) -> Result<Arc<dyn FileOpener>> {
         Ok(Arc::new(CsvOpener {
             config: Arc::new(self.clone()),
             file_compression_type: base_config.file_compression_type,
-            object_store: object_store?,
+            object_store,
         }))
     }
 

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

@@ -162,7 +162,7 @@ impl DataSource for FileScanConfig {
         partition: usize,
         context: Arc<TaskContext>,
     ) -> Result<SendableRecordBatchStream> {
-        let object_store = context.runtime_env().object_store(&self.object_store_url);
+        let object_store = context.runtime_env().object_store(&self.object_store_url)?;
 
         let source = self
             .source

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

@@ -262,7 +262,7 @@ impl JsonSource {
 impl FileSource for JsonSource {
     fn create_file_opener(
         &self,
-        object_store: Result<Arc<dyn ObjectStore>>,
+        object_store: Arc<dyn ObjectStore>,
         base_config: &FileScanConfig,
         _partition: usize,
     ) -> Result<Arc<dyn FileOpener>> {
@@ -272,7 +272,7 @@ impl FileSource for JsonSource {
                 .expect("Batch size must set before creating opener"),
             projected_schema: base_config.projected_file_schema(),
             file_compression_type: base_config.file_compression_type,
-            object_store: object_store?,
+            object_store,
         }))
     }
 

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

@@ -463,7 +463,7 @@ impl ParquetSource {
 impl FileSource for ParquetSource {
     fn create_file_opener(
         &self,
-        object_store: datafusion_common::Result<Arc<dyn ObjectStore>>,
+        object_store: Arc<dyn ObjectStore>,
         base_config: &FileScanConfig,
         partition: usize,
     ) -> datafusion_common::Result<Arc<dyn FileOpener>> {
@@ -475,15 +475,10 @@ impl FileSource for ParquetSource {
             .clone()
             .unwrap_or_else(|| Arc::new(DefaultSchemaAdapterFactory));
 
-        let parquet_file_reader_factory = self
-            .parquet_file_reader_factory
-            .as_ref()
-            .map(|f| Ok(Arc::clone(f)))
-            .unwrap_or_else(|| {
-                object_store.map(|store| {
-                    Arc::new(DefaultParquetFileReaderFactory::new(store)) as _
-                })
-            })?;
+        let parquet_file_reader_factory =
+            self.parquet_file_reader_factory.clone().unwrap_or_else(|| {
+                Arc::new(DefaultParquetFileReaderFactory::new(object_store)) as _
+            });
 
         Ok(Arc::new(ParquetOpener {
             partition_index: partition,
@@ -586,7 +581,4 @@ impl FileSource for ParquetSource {
             }
         }
     }
-    fn supports_repartition(&self, _config: &FileScanConfig) -> bool {
-        true
-    }
 }