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 improvements. The patch includes the following 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 all
  major plans.
- Store `Arc<[usize]>` instead of vector within `FilterExec`.
- Store `Arc<[usize]>` instead of vector within projection of `HashJoinExec`.
- Store `Arc<[Arc<AggregateFunctionExpr>]>` instead of vec for aggr expr and filters.
- Store `Arc<[ProjectionExpr]> instead of vec in `ProjectionExprs` struct.
- Get `Option<&[usize]>` instead of option vec ref in `project_schema` -- it makes API
  more flexible.

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.

Closes #19796

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

@@ -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,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/benches/reset_plan_states.rs

@@ -166,6 +166,8 @@ fn run_reset_states(b: &mut criterion::Bencher, plan: &Arc<dyn ExecutionPlan>) {
 /// making an independent instance of the execution plan to re-execute it, avoiding
 /// re-planning stage.
 fn bench_reset_plan_states(c: &mut Criterion) {
+    env_logger::init();
+
     let rt = Runtime::new().unwrap();
     let ctx = SessionContext::new();
     ctx.register_table(

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