WIP: WorkerExec metadata parallelling ClusterSendExec metadata work in progress

Author: srh

rust/cubestore/cubestore/src/queryplanner/optimizations/distributed_partial_aggregate.rs

@@ -59,13 +59,15 @@ pub fn push_aggregate_to_workers(
     } else if let Some(w) = agg.input().as_any().downcast_ref::<WorkerExec>() {
         let worker_input = p_partial.clone().with_new_children(vec![w.input.clone()])?;
 
+        // TODO upgrade DF: both cs.with_changed_schema and this need required_input_ordering if we have a sorted (inplace) aggregate pair here
+
         // Worker plan, execute partial aggregate inside the worker.
-        Arc::new(WorkerExec {
-            input: worker_input,
-            max_batch_rows: w.max_batch_rows,
-            limit_and_reverse: w.limit_and_reverse.clone(),
-            required_input_ordering: None,
-        })
+        Arc::new(WorkerExec::new(
+            worker_input,
+            w.max_batch_rows,
+            w.limit_and_reverse.clone(),
+            /* required_input_ordering */ None,
+        ))
     } else {
         return Ok(p_final);
     };

rust/cubestore/cubestore/src/queryplanner/panic.rs

@@ -155,10 +155,10 @@ impl ExecutionPlan for PanicWorkerExec {
 }
 
 pub fn plan_panic_worker() -> Result<Arc<dyn ExecutionPlan>, DataFusionError> {
-    Ok(Arc::new(WorkerExec {
-        input: Arc::new(PanicWorkerExec::new()),
-        max_batch_rows: 1,
-        limit_and_reverse: None,
-        required_input_ordering: None,
-    }))
+    Ok(Arc::new(WorkerExec::new(
+        Arc::new(PanicWorkerExec::new()),
+        /* max_batch_rows */ 1,
+        /* limit_and_reverse */ None,
+        /* required_input_ordering */ None,
+    )))
 }

rust/cubestore/cubestore/src/queryplanner/planning.rs

@@ -1719,12 +1719,13 @@ impl CubeExtensionPlanner {
                 required_input_ordering,
             )?))
         } else {
-            Ok(Arc::new(WorkerExec {
+            Ok(Arc::new(WorkerExec::new(
                 input,
                 max_batch_rows,
                 limit_and_reverse,
                 required_input_ordering,
-            }))
+
+            )))
         }
     }
 }
@@ -1737,6 +1738,27 @@ pub struct WorkerExec {
     pub max_batch_rows: usize,
     pub limit_and_reverse: Option<(usize, bool)>,
     pub required_input_ordering: Option<LexRequirement>,
+    properties: PlanProperties,
+}
+
+impl WorkerExec {
+    pub fn new(
+        input: Arc<dyn ExecutionPlan>,
+        max_batch_rows: usize,
+        limit_and_reverse: Option<(usize, bool)>,
+        required_input_ordering: Option<LexRequirement>,
+    ) -> WorkerExec {
+        // TODO upgrade DF: Use partitions_num parameter.
+        let partitions_num = 1;  // TODO upgrade DF: No.
+        let properties = input.properties().clone().with_partitioning(Partitioning::UnknownPartitioning(partitions_num));
+        WorkerExec {
+            input,
+            max_batch_rows,
+            limit_and_reverse,
+            required_input_ordering,
+            properties,
+        }
+    }
 }
 
 impl DisplayAs for WorkerExec {
@@ -1766,6 +1788,7 @@ impl ExecutionPlan for WorkerExec {
             max_batch_rows: self.max_batch_rows,
             limit_and_reverse: self.limit_and_reverse.clone(),
             required_input_ordering: self.required_input_ordering.clone(),
+            properties: self.properties.clone(),
         }))
     }
 
@@ -1782,6 +1805,7 @@ impl ExecutionPlan for WorkerExec {
     }
 
     fn properties(&self) -> &PlanProperties {
+        // TODO upgrade DF: No.
         self.input.properties()
     }
 

rust/cubestore/cubestore/src/queryplanner/query_executor.rs

@@ -1284,7 +1284,7 @@ impl ClusterSendExec {
         })
     }
 
-    fn compute_properties(
+    pub fn compute_properties(
         input_properties: &PlanProperties,
         partitions_num: usize,
     ) -> PlanProperties {