pull schema from FileSource

Author: adriangb

datafusion-examples/examples/advanced_parquet_index.rs

@@ -499,12 +499,11 @@ impl TableProvider for IndexTableProvider {
                 // provide the factory to create parquet reader without re-reading metadata
                 .with_parquet_file_reader_factory(Arc::new(reader_factory)),
         );
-        let file_scan_config =
-            FileScanConfigBuilder::new(object_store_url, schema, file_source)
-                .with_limit(limit)
-                .with_projection_indices(projection.cloned())
-                .with_file(partitioned_file)
-                .build();
+        let file_scan_config = FileScanConfigBuilder::new(object_store_url, file_source)
+            .with_limit(limit)
+            .with_projection_indices(projection.cloned())
+            .with_file(partitioned_file)
+            .build();
 
         // Finally, put it all together into a DataSourceExec
         Ok(DataSourceExec::from_data_source(file_scan_config))

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

@@ -430,9 +430,6 @@ impl TableProvider for ListingTable {
             .map(|col| Ok(Arc::new(self.table_schema.field_with_name(&col.0)?.clone())))
             .collect::<datafusion_common::Result<Vec<_>>>()?;
 
-        let table_schema =
-            TableSchema::new(Arc::clone(&self.file_schema), table_partition_cols.clone());
-
         let table_partition_col_names = table_partition_cols
             .iter()
             .map(|field| field.name().as_str())
@@ -503,19 +500,15 @@ impl TableProvider for ListingTable {
             .format
             .create_physical_plan(
                 state,
-                FileScanConfigBuilder::new(
-                    object_store_url,
-                    table_schema.clone(),
-                    file_source,
-                )
-                .with_file_groups(partitioned_file_lists)
-                .with_constraints(self.constraints.clone())
-                .with_statistics(statistics)
-                .with_projection_indices(projection)
-                .with_limit(limit)
-                .with_output_ordering(output_ordering)
-                .with_expr_adapter(self.expr_adapter_factory.clone())
-                .build(),
+                FileScanConfigBuilder::new(object_store_url, file_source)
+                    .with_file_groups(partitioned_file_lists)
+                    .with_constraints(self.constraints.clone())
+                    .with_statistics(statistics)
+                    .with_projection_indices(projection)
+                    .with_limit(limit)
+                    .with_output_ordering(output_ordering)
+                    .with_expr_adapter(self.expr_adapter_factory.clone())
+                    .build(),
             )
             .await?;
 

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

@@ -127,13 +127,10 @@ mod tests {
         let source = ParquetSource::new(Arc::clone(&schema))
             .with_schema_adapter_factory(Arc::new(TestSchemaAdapterFactory {}))
             .unwrap();
-        let base_conf = FileScanConfigBuilder::new(
-            ObjectStoreUrl::local_filesystem(),
-            schema,
-            source,
-        )
-        .with_file(partitioned_file)
-        .build();
+        let base_conf =
+            FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), source)
+                .with_file(partitioned_file)
+                .build();
 
         let parquet_exec = DataSourceExec::from_data_source(base_conf);
 

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

@@ -82,14 +82,10 @@ mod tests {
             .await?;
 
         let source = Arc::new(AvroSource::new(Arc::clone(&file_schema)));
-        let conf = FileScanConfigBuilder::new(
-            ObjectStoreUrl::local_filesystem(),
-            file_schema,
-            source,
-        )
-        .with_file(meta.into())
-        .with_projection_indices(Some(vec![0, 1, 2]))
-        .build();
+        let conf = FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), source)
+            .with_file(meta.into())
+            .with_projection_indices(Some(vec![0, 1, 2]))
+            .build();
 
         let source_exec = DataSourceExec::from_data_source(conf);
         assert_eq!(
@@ -158,7 +154,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 +223,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

@@ -123,15 +123,12 @@ mod tests {
         let table_schema = TableSchema::from_file_schema(Arc::clone(&file_schema));
         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,
-        ))
-        .with_file_compression_type(file_compression_type)
-        .with_newlines_in_values(false)
-        .with_projection_indices(Some(vec![0, 2, 4]))
-        .build();
+        let config =
+            FileScanConfigBuilder::from(partitioned_csv_config(file_groups, source))
+                .with_file_compression_type(file_compression_type)
+                .with_newlines_in_values(false)
+                .with_projection_indices(Some(vec![0, 2, 4]))
+                .build();
 
         assert_eq!(13, config.file_schema().fields().len());
         let csv = DataSourceExec::from_data_source(config);
@@ -198,15 +195,12 @@ mod tests {
         let table_schema = TableSchema::from_file_schema(Arc::clone(&file_schema));
         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,
-        ))
-        .with_newlines_in_values(false)
-        .with_file_compression_type(file_compression_type.to_owned())
-        .with_projection_indices(Some(vec![4, 0, 2]))
-        .build();
+        let config =
+            FileScanConfigBuilder::from(partitioned_csv_config(file_groups, source))
+                .with_newlines_in_values(false)
+                .with_file_compression_type(file_compression_type.to_owned())
+                .with_projection_indices(Some(vec![4, 0, 2]))
+                .build();
         assert_eq!(13, config.file_schema().fields().len());
         let csv = DataSourceExec::from_data_source(config);
         assert_eq!(3, csv.schema().fields().len());
@@ -272,15 +266,12 @@ mod tests {
         let table_schema = TableSchema::from_file_schema(Arc::clone(&file_schema));
         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,
-        ))
-        .with_newlines_in_values(false)
-        .with_file_compression_type(file_compression_type.to_owned())
-        .with_limit(Some(5))
-        .build();
+        let config =
+            FileScanConfigBuilder::from(partitioned_csv_config(file_groups, source))
+                .with_newlines_in_values(false)
+                .with_file_compression_type(file_compression_type.to_owned())
+                .with_limit(Some(5))
+                .build();
         assert_eq!(13, config.file_schema().fields().len());
         let csv = DataSourceExec::from_data_source(config);
         assert_eq!(13, csv.schema().fields().len());
@@ -345,15 +336,12 @@ mod tests {
         let table_schema = TableSchema::from_file_schema(Arc::clone(&file_schema));
         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,
-        ))
-        .with_newlines_in_values(false)
-        .with_file_compression_type(file_compression_type.to_owned())
-        .with_limit(Some(5))
-        .build();
+        let config =
+            FileScanConfigBuilder::from(partitioned_csv_config(file_groups, source))
+                .with_newlines_in_values(false)
+                .with_file_compression_type(file_compression_type.to_owned())
+                .with_limit(Some(5))
+                .build();
         assert_eq!(14, config.file_schema().fields().len());
         let csv = DataSourceExec::from_data_source(config);
         assert_eq!(14, csv.schema().fields().len());
@@ -410,23 +398,20 @@ 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 config = FileScanConfigBuilder::from(partitioned_csv_config(
-            table_schema,
-            file_groups,
-            source,
-        ))
-        .with_newlines_in_values(false)
-        .with_file_compression_type(file_compression_type.to_owned())
-        // We should be able to project on the partition column
-        // Which is supposed to be after the file fields
-        .with_projection_indices(Some(vec![0, num_file_schema_fields]))
-        .build();
+        let source =
+            Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options));
+        let config =
+            FileScanConfigBuilder::from(partitioned_csv_config(file_groups, source))
+                .with_newlines_in_values(false)
+                .with_file_compression_type(file_compression_type.to_owned())
+                // We should be able to project on the partition column
+                // Which is supposed to be after the file fields
+                .with_projection_indices(Some(vec![0, num_file_schema_fields]))
+                .build();
 
         // we don't have `/date=xx/` in the path but that is ok because
         // partitions are resolved during scan anyway
@@ -524,14 +509,11 @@ mod tests {
         let table_schema = TableSchema::from_file_schema(Arc::clone(&file_schema));
         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,
-        ))
-        .with_newlines_in_values(false)
-        .with_file_compression_type(file_compression_type.to_owned())
-        .build();
+        let config =
+            FileScanConfigBuilder::from(partitioned_csv_config(file_groups, source))
+                .with_newlines_in_values(false)
+                .with_file_compression_type(file_compression_type.to_owned())
+                .build();
         let csv = DataSourceExec::from_data_source(config);
 
         let it = csv.execute(0, task_ctx).unwrap();