add fast-path for with_new_children

This patch aims to implement a fast-path for the ExecutionPlan::with_new_children function
for some plans, moving closer to a physical plan re-use implementation and improving planning
performance. If the passed children properties are the same as in self, we do not actually
recompute self's properties  (which could be costly if projection mapping is required).
Instead, we just replace the children and re-use self's properties as-is.

To be able to compare two different properties -- ExecutionPlan::properties(...) signature
is modified and now returns `&Arc<PlanProperties>`. If `children` properties are the same
in `with_new_children` -- we clone our properties arc and then a parent plan will consider
our properties as unchanged, doing the same.

Also, there are other improvenets, all changes:

- Return `&Arc<PlanProperties>` from `ExecutionPlan::properties(...)` instead of a reference.
- Implement `with_new_children` fast-path if there is no children properties changes for plans:
  * SortExec
  * RepartitionExec
  * ProjectionExec
  * FilterExec
  * CoalescePartitionsExec
  * AggregateExec
- Export `reset_plan_states` function.
- Store `Arc<[usize]>` instead of vector within `FilterExec`.
- Store `Arc<[Arc<AggregateFunctionExpr>]>` instead of vec for aggr expr and filters.
- Store `Arc<[ProjectionExpr]> instead of vec in `ProjectionExprs` struct.

Note: currently, `reset_plan_states` does not allow to re-use plan in general: it is not
supported for dynamic filters and recursive queries features, as in this case state reset
should update pointers in the children plans.

Author: askalt

datafusion-examples/examples/custom_data_source/custom_datasource.rs

@@ -192,7 +192,7 @@ impl TableProvider for CustomDataSource {
 struct CustomExec {
     db: CustomDataSource,
     projected_schema: SchemaRef,
-    cache: PlanProperties,
+    cache: Arc<PlanProperties>,
 }
 
 impl CustomExec {
@@ -202,12 +202,13 @@ impl CustomExec {
         schema: SchemaRef,
         db: CustomDataSource,
     ) -> Self {
-        let projected_schema = project_schema(&schema, projections).unwrap();
+        let projected_schema =
+            project_schema(&schema, projections.map(AsRef::as_ref)).unwrap();
         let cache = Self::compute_properties(projected_schema.clone());
         Self {
             db,
             projected_schema,
-            cache,
+            cache: Arc::new(cache),
         }
     }
 
@@ -238,7 +239,7 @@ impl ExecutionPlan for CustomExec {
         self
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.cache
     }
 

datafusion-examples/examples/execution_monitoring/memory_pool_execution_plan.rs

@@ -199,7 +199,7 @@ impl ExternalBatchBufferer {
 struct BufferingExecutionPlan {
     schema: SchemaRef,
     input: Arc<dyn ExecutionPlan>,
-    properties: PlanProperties,
+    properties: Arc<PlanProperties>,
 }
 
 impl BufferingExecutionPlan {
@@ -233,7 +233,7 @@ impl ExecutionPlan for BufferingExecutionPlan {
         self.schema.clone()
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }
 

datafusion-examples/examples/proto/composed_extension_codec.rs

@@ -106,7 +106,7 @@ impl ExecutionPlan for ParentExec {
         self
     }
 
-    fn properties(&self) -> &datafusion::physical_plan::PlanProperties {
+    fn properties(&self) -> &Arc<datafusion::physical_plan::PlanProperties> {
         unreachable!()
     }
 
@@ -182,7 +182,7 @@ impl ExecutionPlan for ChildExec {
         self
     }
 
-    fn properties(&self) -> &datafusion::physical_plan::PlanProperties {
+    fn properties(&self) -> &Arc<datafusion::physical_plan::PlanProperties> {
         unreachable!()
     }
 

datafusion-examples/examples/relation_planner/table_sample.rs

@@ -618,7 +618,7 @@ pub struct SampleExec {
     upper_bound: f64,
     seed: u64,
     metrics: ExecutionPlanMetricsSet,
-    cache: PlanProperties,
+    cache: Arc<PlanProperties>,
 }
 
 impl SampleExec {
@@ -656,7 +656,7 @@ impl SampleExec {
             upper_bound,
             seed,
             metrics: ExecutionPlanMetricsSet::new(),
-            cache,
+            cache: Arc::new(cache),
         })
     }
 
@@ -686,7 +686,7 @@ impl ExecutionPlan for SampleExec {
         self
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.cache
     }
 

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/catalog/src/memory/table.rs

@@ -549,7 +549,7 @@ fn evaluate_filters_to_mask(
 struct DmlResultExec {
     rows_affected: u64,
     schema: SchemaRef,
-    properties: PlanProperties,
+    properties: Arc<PlanProperties>,
 }
 
 impl DmlResultExec {
@@ -570,7 +570,7 @@ impl DmlResultExec {
         Self {
             rows_affected,
             schema,
-            properties,
+            properties: Arc::new(properties),
         }
     }
 }
@@ -604,7 +604,7 @@ impl ExecutionPlan for DmlResultExec {
         Arc::clone(&self.schema)
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.properties
     }
 

datafusion/common/src/stats.rs

@@ -391,7 +391,7 @@ 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 {
+    pub fn project(mut self, projection: Option<&[usize]>) -> Self {
         let Some(projection) = projection else {
             return self;
         };
@@ -1066,29 +1066,28 @@ mod tests {
 
     #[test]
     fn test_project_none() {
-        let projection = None;
-        let stats = make_stats(vec![10, 20, 30]).project(projection.as_ref());
+        let stats = make_stats(vec![10, 20, 30]).project(None);
         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 projection: Option<&[_]> = Some(&[]);
+        let stats = make_stats(vec![10, 20, 30]).project(projection);
         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 projection: Option<&[_]> = Some(&[2, 1]);
+        let stats = make_stats(vec![10, 20, 30]).project(projection);
         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 projection: Option<&[_]> = Some(&[1, 2, 1, 1, 0, 2]);
+        let stats = make_stats(vec![10, 20, 30]).project(projection);
         assert_eq!(stats, make_stats(vec![20, 30, 20, 20, 10, 30]));
     }
 

datafusion/common/src/utils/mod.rs

@@ -70,7 +70,7 @@ use std::thread::available_parallelism;
 /// ```
 pub fn project_schema(
     schema: &SchemaRef,
-    projection: Option<&Vec<usize>>,
+    projection: Option<&[usize]>,
 ) -> Result<SchemaRef> {
     let schema = match projection {
         Some(columns) => Arc::new(schema.project(columns)?),

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

@@ -3661,13 +3661,15 @@ mod tests {
 
     #[derive(Debug)]
     struct NoOpExecutionPlan {
-        cache: PlanProperties,
+        cache: Arc<PlanProperties>,
     }
 
     impl NoOpExecutionPlan {
         fn new(schema: SchemaRef) -> Self {
             let cache = Self::compute_properties(schema);
-            Self { cache }
+            Self {
+                cache: Arc::new(cache),
+            }
         }
 
         /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
@@ -3705,7 +3707,7 @@ mod tests {
             self
         }
 
-        fn properties(&self) -> &PlanProperties {
+        fn properties(&self) -> &Arc<PlanProperties> {
             &self.cache
         }
 
@@ -3859,7 +3861,7 @@ digraph {
         fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
             self.0.iter().collect::<Vec<_>>()
         }
-        fn properties(&self) -> &PlanProperties {
+        fn properties(&self) -> &Arc<PlanProperties> {
             unimplemented!()
         }
         fn execute(
@@ -3908,7 +3910,7 @@ digraph {
         fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
             unimplemented!()
         }
-        fn properties(&self) -> &PlanProperties {
+        fn properties(&self) -> &Arc<PlanProperties> {
             unimplemented!()
         }
         fn execute(
@@ -4029,7 +4031,7 @@ digraph {
         fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
             vec![]
         }
-        fn properties(&self) -> &PlanProperties {
+        fn properties(&self) -> &Arc<PlanProperties> {
             unimplemented!()
         }
         fn execute(