Let FileScanConfig own a list of ProjectionExprs (#18253)

## Which issue does this PR close?

- Related to #14993

## Rationale for this change

To enable expression pushdown to file sources, we need to plumb
expressions through the `FileScanConfig` layer. Currently,
`FileScanConfig` only tracks column indices for projection, which limits
us to simple and naive column selection.

This PR begins expression pushdown implementation by having
`FileScanConfig` own a list of `ProjectionExpr`s, instead of column
indices. This allows file sources to eventually receive and optimize
based on the actual expressions being projected.


## Notes about this PR
- The first commit is based off of
#18231
- To avoid a super large diff and a harder review, I've decided to break
(#14993) into 2 tasks:
- Have the `DataSource` (`FileScanConfig`) actually hold projection
expressions (this PR)
- Flow the projection expressions from `DataSourceExec` all the way to
the `FileSource`

---------

Co-authored-by: Adrian Garcia Badaracco <[email protected]>

Author: friendlymatthew

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),