Let FileScanConfig own a list of ProjectionExprs

datafusion-examples/examples/advanced_parquet_index.rs

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

datafusion-examples/examples/csv_json_opener.rs

@@ -60,7 +60,7 @@ async fn csv_opener() -> Result<()> {
         Arc::clone(&schema),
         Arc::new(CsvSource::default()),
     )
-    .with_projection(Some(vec![12, 0]))
+    .with_projection_indices(Some(vec![12, 0]))
     .with_limit(Some(5))
     .with_file(PartitionedFile::new(path.display().to_string(), 10))
     .build();
@@ -126,7 +126,7 @@ async fn json_opener() -> Result<()> {
         schema,
         Arc::new(JsonSource::default()),
     )
-    .with_projection(Some(vec![1, 0]))
+    .with_projection_indices(Some(vec![1, 0]))
     .with_limit(Some(5))
     .with_file(PartitionedFile::new(path.to_string(), 10))
     .build();

datafusion-examples/examples/default_column_values.rs

@@ -260,7 +260,7 @@ impl TableProvider for DefaultValueTableProvider {
             self.schema.clone(),
             Arc::new(parquet_source),
         )
-        .with_projection(projection.cloned())
+        .with_projection_indices(projection.cloned())
         .with_limit(limit)
         .with_file_group(file_group)
         .with_expr_adapter(Some(Arc::new(DefaultValuePhysicalExprAdapterFactory) as _));

datafusion-examples/examples/parquet_index.rs

@@ -246,7 +246,7 @@ impl TableProvider for IndexTableProvider {
         let source = Arc::new(ParquetSource::default().with_predicate(predicate));
         let mut file_scan_config_builder =
             FileScanConfigBuilder::new(object_store_url, self.schema(), source)
-                .with_projection(projection.cloned())
+                .with_projection_indices(projection.cloned())
                 .with_limit(limit);
 
         // Transform to the format needed to pass to DataSourceExec

datafusion/catalog-listing/src/table.rs

@@ -499,7 +499,7 @@ impl TableProvider for ListingTable {
                 .with_file_groups(partitioned_file_lists)
                 .with_constraints(self.constraints.clone())
                 .with_statistics(statistics)
-                .with_projection(projection)
+                .with_projection_indices(projection)
                 .with_limit(limit)
                 .with_output_ordering(output_ordering)
                 .with_table_partition_cols(table_partition_cols)

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

@@ -90,7 +90,7 @@ pub(crate) mod test_util {
                 )
                 .with_file_groups(file_groups)
                 .with_statistics(statistics)
-                .with_projection(projection)
+                .with_projection_indices(projection)
                 .with_limit(limit)
                 .build(),
             )

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

@@ -88,7 +88,7 @@ mod tests {
             source,
         )
         .with_file(meta.into())
-        .with_projection(Some(vec![0, 1, 2]))
+        .with_projection_indices(Some(vec![0, 1, 2]))
         .build();
 
         let source_exec = DataSourceExec::from_data_source(conf);
@@ -160,7 +160,7 @@ mod tests {
         let source = Arc::new(AvroSource::new());
         let conf = FileScanConfigBuilder::new(object_store_url, file_schema, source)
             .with_file(meta.into())
-            .with_projection(projection)
+            .with_projection_indices(projection)
             .build();
 
         let source_exec = DataSourceExec::from_data_source(conf);
@@ -231,7 +231,7 @@ mod tests {
         let conf = FileScanConfigBuilder::new(object_store_url, file_schema, 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(projection)
+            .with_projection_indices(projection)
             .with_file(partitioned_file)
             .with_table_partition_cols(vec![Field::new("date", DataType::Utf8, false)])
             .build();

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

@@ -118,7 +118,7 @@ mod tests {
         ))
         .with_file_compression_type(file_compression_type)
         .with_newlines_in_values(false)
-        .with_projection(Some(vec![0, 2, 4]))
+        .with_projection_indices(Some(vec![0, 2, 4]))
         .build();
 
         assert_eq!(13, config.file_schema().fields().len());
@@ -183,7 +183,7 @@ mod tests {
         ))
         .with_newlines_in_values(false)
         .with_file_compression_type(file_compression_type.to_owned())
-        .with_projection(Some(vec![4, 0, 2]))
+        .with_projection_indices(Some(vec![4, 0, 2]))
         .build();
         assert_eq!(13, config.file_schema().fields().len());
         let csv = DataSourceExec::from_data_source(config);
@@ -373,7 +373,7 @@ mod tests {
         .with_table_partition_cols(vec![Field::new("date", DataType::Utf8, false)])
         // We should be able to project on the partition column
         // Which is supposed to be after the file fields
-        .with_projection(Some(vec![0, num_file_schema_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

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

@@ -297,7 +297,7 @@ mod tests {
         let source = Arc::new(JsonSource::new());
         let conf = FileScanConfigBuilder::new(object_store_url, file_schema, source)
             .with_file_groups(file_groups)
-            .with_projection(Some(vec![0, 2]))
+            .with_projection_indices(Some(vec![0, 2]))
             .with_file_compression_type(file_compression_type.to_owned())
             .build();
         let exec = DataSourceExec::from_data_source(conf);
@@ -345,7 +345,7 @@ mod tests {
         let source = Arc::new(JsonSource::new());
         let conf = FileScanConfigBuilder::new(object_store_url, file_schema, source)
             .with_file_groups(file_groups)
-            .with_projection(Some(vec![3, 0, 2]))
+            .with_projection_indices(Some(vec![3, 0, 2]))
             .with_file_compression_type(file_compression_type.to_owned())
             .build();
         let exec = DataSourceExec::from_data_source(conf);

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

@@ -201,7 +201,7 @@ mod tests {
                 source,
             )
             .with_file_group(file_group)
-            .with_projection(self.projection.clone())
+            .with_projection_indices(self.projection.clone())
             .build();
             DataSourceExec::from_data_source(base_config)
         }
@@ -1655,7 +1655,7 @@ mod tests {
         let config = FileScanConfigBuilder::new(object_store_url, schema.clone(), source)
             .with_file(partitioned_file)
             // file has 10 cols so index 12 should be month and 13 should be day
-            .with_projection(Some(vec![0, 1, 2, 12, 13]))
+            .with_projection_indices(Some(vec![0, 1, 2, 12, 13]))
             .with_table_partition_cols(vec![
                 Field::new("year", DataType::Utf8, false),
                 Field::new("month", DataType::UInt8, false),

datafusion/core/tests/parquet/schema_coercion.rs

@@ -126,7 +126,7 @@ async fn multi_parquet_coercion_projection() {
         Arc::new(ParquetSource::default()),
     )
     .with_file_group(file_group)
-    .with_projection(Some(vec![1, 0, 2]))
+    .with_projection_indices(Some(vec![1, 0, 2]))
     .build();
 
     let parquet_exec = DataSourceExec::from_data_source(config);

datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs

@@ -165,7 +165,7 @@ impl FileSource for TestSource {
 
     fn with_projection(&self, config: &FileScanConfig) -> Arc<dyn FileSource> {
         Arc::new(TestSource {
-            projection: config.projection.clone(),
+            projection: config.projection_exprs.as_ref().map(|p| p.column_indices()),
             ..self.clone()
         })
     }

datafusion/core/tests/physical_optimizer/projection_pushdown.rs

@@ -390,7 +390,7 @@ fn create_simple_csv_exec() -> Arc<dyn ExecutionPlan> {
         Arc::new(CsvSource::new(false, 0, 0)),
     )
     .with_file(PartitionedFile::new("x".to_string(), 100))
-    .with_projection(Some(vec![0, 1, 2, 3, 4]))
+    .with_projection_indices(Some(vec![0, 1, 2, 3, 4]))
     .build();
 
     DataSourceExec::from_data_source(config)
@@ -409,7 +409,7 @@ fn create_projecting_csv_exec() -> Arc<dyn ExecutionPlan> {
         Arc::new(CsvSource::new(false, 0, 0)),
     )
     .with_file(PartitionedFile::new("x".to_string(), 100))
-    .with_projection(Some(vec![3, 2, 1]))
+    .with_projection_indices(Some(vec![3, 2, 1]))
     .build();
 
     DataSourceExec::from_data_source(config)
@@ -1596,7 +1596,7 @@ fn partitioned_data_source() -> Arc<DataSourceExec> {
     )
     .with_file(PartitionedFile::new("x".to_string(), 100))
     .with_table_partition_cols(vec![Field::new("partition_col", DataType::Utf8, true)])
-    .with_projection(Some(vec![0, 1, 2]))
+    .with_projection_indices(Some(vec![0, 1, 2]))
     .build();
 
     DataSourceExec::from_data_source(config)