pull schema from FileSource

Author: adriangb

datafusion-examples/examples/advanced_parquet_index.rs

@@ -500,7 +500,7 @@ impl TableProvider for IndexTableProvider {
                 .with_parquet_file_reader_factory(Arc::new(reader_factory)),
         );
         let file_scan_config =
-            FileScanConfigBuilder::new(object_store_url, schema, file_source)
+            FileScanConfigBuilder::new(object_store_url, file_source)
                 .with_limit(limit)
                 .with_projection_indices(projection.cloned())
                 .with_file(partitioned_file)

datafusion-examples/examples/csv_json_opener.rs

@@ -65,7 +65,6 @@ async fn csv_opener() -> Result<()> {
 
     let scan_config = FileScanConfigBuilder::new(
         ObjectStoreUrl::local_filesystem(),
-        Arc::clone(&schema),
         Arc::new(CsvSource::new(Arc::clone(&schema)).with_csv_options(options.clone())),
     )
     .with_projection_indices(Some(vec![12, 0]))
@@ -131,8 +130,7 @@ async fn json_opener() -> Result<()> {
 
     let scan_config = FileScanConfigBuilder::new(
         ObjectStoreUrl::local_filesystem(),
-        schema,
-        Arc::new(JsonSource::default()),
+        Arc::new(JsonSource::new(schema)),
     )
     .with_projection_indices(Some(vec![1, 0]))
     .with_limit(Some(5))

datafusion-examples/examples/default_column_values.rs

@@ -257,7 +257,6 @@ impl TableProvider for DefaultValueTableProvider {
 
         let file_scan_config = FileScanConfigBuilder::new(
             ObjectStoreUrl::parse("memory://")?,
-            self.schema.clone(),
             Arc::new(parquet_source),
         )
         .with_projection_indices(projection.cloned())

datafusion-examples/examples/parquet_embedded_index.rs

@@ -429,7 +429,7 @@ impl TableProvider for DistinctIndexTable {
         let source = Arc::new(
             ParquetSource::new(self.schema.clone()).with_enable_page_index(true),
         );
-        let mut builder = FileScanConfigBuilder::new(url, self.schema.clone(), source);
+        let mut builder = FileScanConfigBuilder::new(url, source);
         for file in files_to_scan {
             let path = self.dir.join(file);
             let len = std::fs::metadata(&path)?.len();

datafusion-examples/examples/parquet_index.rs

@@ -246,7 +246,7 @@ impl TableProvider for IndexTableProvider {
         let source =
             Arc::new(ParquetSource::new(self.schema()).with_predicate(predicate));
         let mut file_scan_config_builder =
-            FileScanConfigBuilder::new(object_store_url, self.schema(), source)
+            FileScanConfigBuilder::new(object_store_url, source)
                 .with_projection_indices(projection.cloned())
                 .with_limit(limit);
 

datafusion/catalog-listing/src/table.rs

@@ -505,7 +505,6 @@ impl TableProvider for ListingTable {
                 state,
                 FileScanConfigBuilder::new(
                     object_store_url,
-                    table_schema.clone(),
                     file_source,
                 )
                 .with_file_groups(partitioned_file_lists)

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

@@ -88,7 +88,6 @@ pub(crate) mod test_util {
                 state,
                 FileScanConfigBuilder::new(
                     ObjectStoreUrl::local_filesystem(),
-                    file_schema.clone(),
                     format.file_source(table_schema),
                 )
                 .with_file_groups(file_groups)

datafusion/core/src/datasource/mod.rs

@@ -129,7 +129,6 @@ mod tests {
             .unwrap();
         let base_conf = FileScanConfigBuilder::new(
             ObjectStoreUrl::local_filesystem(),
-            schema,
             source,
         )
         .with_file(partitioned_file)

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

@@ -84,7 +84,6 @@ mod tests {
         let source = Arc::new(AvroSource::new(Arc::clone(&file_schema)));
         let conf = FileScanConfigBuilder::new(
             ObjectStoreUrl::local_filesystem(),
-            file_schema,
             source,
         )
         .with_file(meta.into())
@@ -158,7 +157,7 @@ mod tests {
         let projection = Some(vec![0, 1, 2, actual_schema.fields().len()]);
 
         let source = Arc::new(AvroSource::new(Arc::clone(&file_schema)));
-        let conf = FileScanConfigBuilder::new(object_store_url, file_schema, source)
+        let conf = FileScanConfigBuilder::new(object_store_url, source)
             .with_file(meta.into())
             .with_projection_indices(projection)
             .build();
@@ -227,12 +226,12 @@ mod tests {
         partitioned_file.partition_values = vec![ScalarValue::from("2021-10-26")];
 
         let projection = Some(vec![0, 1, file_schema.fields().len(), 2]);
-        let source = Arc::new(AvroSource::new(Arc::clone(&file_schema)));
         let table_schema = TableSchema::new(
             file_schema.clone(),
             vec![Arc::new(Field::new("date", DataType::Utf8, false))],
         );
-        let conf = FileScanConfigBuilder::new(object_store_url, table_schema, source)
+        let source = Arc::new(AvroSource::new(table_schema.clone()));
+        let conf = FileScanConfigBuilder::new(object_store_url, source)
             // select specific columns of the files as well as the partitioning
             // column which is supposed to be the last column in the table schema.
             .with_projection_indices(projection)

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

@@ -124,7 +124,6 @@ mod tests {
         let source =
             Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options));
         let config = FileScanConfigBuilder::from(partitioned_csv_config(
-            table_schema,
             file_groups,
             source,
         ))
@@ -199,7 +198,6 @@ mod tests {
         let source =
             Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options));
         let config = FileScanConfigBuilder::from(partitioned_csv_config(
-            table_schema,
             file_groups,
             source,
         ))
@@ -273,7 +271,6 @@ mod tests {
         let source =
             Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options));
         let config = FileScanConfigBuilder::from(partitioned_csv_config(
-            table_schema,
             file_groups,
             source,
         ))
@@ -346,7 +343,6 @@ mod tests {
         let source =
             Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options));
         let config = FileScanConfigBuilder::from(partitioned_csv_config(
-            table_schema,
             file_groups,
             source,
         ))
@@ -410,14 +406,13 @@ mod tests {
             quote: b'"',
             ..Default::default()
         };
-        let source =
-            Arc::new(CsvSource::new(Arc::clone(&file_schema)).with_csv_options(options));
         let table_schema = TableSchema::new(
             Arc::clone(&file_schema),
             vec![Arc::new(Field::new("date", DataType::Utf8, false))],
         );
+        let source =
+            Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options));
         let config = FileScanConfigBuilder::from(partitioned_csv_config(
-            table_schema,
             file_groups,
             source,
         ))
@@ -525,7 +520,6 @@ mod tests {
         let source =
             Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options));
         let config = FileScanConfigBuilder::from(partitioned_csv_config(
-            table_schema,
             file_groups,
             source,
         ))