Wrap immutable plan parts into Arc

- Closes #19852

Improve performance of query planning and plan state re-set by making node clone cheap.

- Store projection as `Option<Arc<[usize]>>` instead of `Option<Vec<usize>>` in `FilterExec`,
  `HashJoinExec`, `NestedLoopJoinExec`.
- Store exprs as `Arc<[ProjectionExpr]>` instead of Vec in `ProjectionExprs`.
- Store arced aggregation, filter, group by expressions within `AggregateExec`.

Author: askalt

datafusion-examples/examples/custom_data_source/custom_datasource.rs

@@ -202,7 +202,8 @@ impl CustomExec {
         schema: SchemaRef,
         db: CustomDataSource,
     ) -> Self {
-        let projected_schema = project_schema(&schema, projections).unwrap();
+        let projected_schema =
+            project_schema(&schema, projections.map(|v| v.as_ref())).unwrap();
         let cache = Self::compute_properties(projected_schema.clone());
         Self {
             db,

datafusion/catalog-listing/src/table.rs

@@ -522,7 +522,7 @@ impl TableProvider for ListingTable {
 
         // if no files need to be read, return an `EmptyExec`
         if partitioned_file_lists.is_empty() {
-            let projected_schema = project_schema(&self.schema(), projection.as_ref())?;
+            let projected_schema = project_schema(&self.schema(), projection.as_deref())?;
             return Ok(ScanResult::new(Arc::new(EmptyExec::new(projected_schema))));
         }
 

datafusion/common/src/stats.rs

@@ -391,8 +391,8 @@ impl Statistics {
     /// For example, if we had statistics for columns `{"a", "b", "c"}`,
     /// projecting to `vec![2, 1]` would return statistics for columns `{"c",
     /// "b"}`.
-    pub fn project(mut self, projection: Option<&Vec<usize>>) -> Self {
-        let Some(projection) = projection else {
+    pub fn project(mut self, projection: Option<&[usize]>) -> Self {
+        let Some(projection) = projection.map(AsRef::as_ref) else {
             return self;
         };
 
@@ -410,7 +410,7 @@ impl Statistics {
             .map(Slot::Present)
             .collect();
 
-        for idx in projection {
+        for idx in projection.iter() {
             let next_idx = self.column_statistics.len();
             let slot = std::mem::replace(
                 columns.get_mut(*idx).expect("projection out of bounds"),
@@ -1066,29 +1066,29 @@ mod tests {
 
     #[test]
     fn test_project_none() {
-        let projection = None;
-        let stats = make_stats(vec![10, 20, 30]).project(projection.as_ref());
+        let projection: Option<&[usize]> = None;
+        let stats = make_stats(vec![10, 20, 30]).project(projection);
         assert_eq!(stats, make_stats(vec![10, 20, 30]));
     }
 
     #[test]
     fn test_project_empty() {
         let projection = Some(vec![]);
-        let stats = make_stats(vec![10, 20, 30]).project(projection.as_ref());
+        let stats = make_stats(vec![10, 20, 30]).project(projection.as_deref());
         assert_eq!(stats, make_stats(vec![]));
     }
 
     #[test]
     fn test_project_swap() {
         let projection = Some(vec![2, 1]);
-        let stats = make_stats(vec![10, 20, 30]).project(projection.as_ref());
+        let stats = make_stats(vec![10, 20, 30]).project(projection.as_deref());
         assert_eq!(stats, make_stats(vec![30, 20]));
     }
 
     #[test]
     fn test_project_repeated() {
         let projection = Some(vec![1, 2, 1, 1, 0, 2]);
-        let stats = make_stats(vec![10, 20, 30]).project(projection.as_ref());
+        let stats = make_stats(vec![10, 20, 30]).project(projection.as_deref());
         assert_eq!(stats, make_stats(vec![20, 30, 20, 20, 10, 30]));
     }
 

datafusion/common/src/utils/mod.rs

@@ -59,7 +59,7 @@ use std::thread::available_parallelism;
 ///
 /// // Pick columns 'c' and 'b'
 /// let projection = Some(vec![2, 1]);
-/// let projected_schema = project_schema(&schema, projection.as_ref()).unwrap();
+/// let projected_schema = project_schema(&schema, projection.as_deref()).unwrap();
 ///
 /// let expected_schema = SchemaRef::new(Schema::new(vec![
 ///     Field::new("c", DataType::Utf8, true),
@@ -70,9 +70,9 @@ use std::thread::available_parallelism;
 /// ```
 pub fn project_schema(
     schema: &SchemaRef,
-    projection: Option<&Vec<usize>>,
+    projection: Option<&[usize]>,
 ) -> Result<SchemaRef> {
-    let schema = match projection {
+    let schema = match projection.map(AsRef::as_ref) {
         Some(columns) => Arc::new(schema.project(columns)?),
         None => Arc::clone(schema),
     };

datafusion/core/src/datasource/empty.rs

@@ -77,7 +77,8 @@ impl TableProvider for EmptyTable {
         _limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         // even though there is no data, projections apply
-        let projected_schema = project_schema(&self.schema, projection)?;
+        let projected_schema =
+            project_schema(&self.schema, projection.map(AsRef::as_ref))?;
         Ok(Arc::new(
             EmptyExec::new(projected_schema).with_partitions(self.partitions),
         ))

datafusion/core/src/physical_planner.rs

@@ -984,7 +984,7 @@ impl DefaultPhysicalPlanner {
                         // project the output columns excluding the async functions
                         // The async functions are always appended to the end of the schema.
                         .apply_projection(Some(
-                            (0..input.schema().fields().len()).collect(),
+                            (0..input.schema().fields().len()).collect::<Vec<_>>(),
                         ))?
                         .with_batch_size(session_state.config().batch_size())
                         .build()?

datafusion/core/tests/custom_sources_cases/mod.rs

@@ -86,7 +86,7 @@ impl CustomExecutionPlan {
     fn new(projection: Option<Vec<usize>>) -> Self {
         let schema = TEST_CUSTOM_SCHEMA_REF!();
         let schema =
-            project_schema(&schema, projection.as_ref()).expect("projected schema");
+            project_schema(&schema, projection.as_deref()).expect("projected schema");
         let cache = Self::compute_properties(schema);
         Self { projection, cache }
     }

datafusion/core/tests/physical_optimizer/join_selection.rs

@@ -762,7 +762,7 @@ async fn test_hash_join_swap_on_joins_with_projections(
             "ProjectionExec won't be added above if HashJoinExec contains embedded projection",
         );
 
-    assert_eq!(swapped_join.projection, Some(vec![0_usize]));
+    assert_eq!(swapped_join.projection.as_ref().unwrap(), [0_usize]);
     assert_eq!(swapped.schema().fields.len(), 1);
     assert_eq!(swapped.schema().fields[0].name(), "small_col");
     Ok(())

datafusion/datasource/src/memory.rs

@@ -262,7 +262,7 @@ impl MemorySourceConfig {
         schema: SchemaRef,
         projection: Option<Vec<usize>>,
     ) -> Result<Self> {
-        let projected_schema = project_schema(&schema, projection.as_ref())?;
+        let projected_schema = project_schema(&schema, projection.as_deref())?;
         Ok(Self {
             partitions: partitions.to_vec(),
             schema,