chore(cubestore): Upgrade DF: Reduce the amount of redundant planning and optimization

Author: srh

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

@@ -313,16 +313,13 @@ impl QueryExecutor for QueryExecutorImpl {
         )?;
         let pre_serialized_plan = Arc::new(pre_serialized_plan);
         let ctx = self.router_context(cluster.clone(), pre_serialized_plan.clone())?;
-        let router_plan = ctx
-            .clone()
-            .state()
-            .create_physical_plan(pre_serialized_plan.logical_plan())
-            .await?;
+        // We don't want to use session_state.create_physical_plan(...) because it redundantly
+        // optimizes the logical plan, which has already been optimized before it was put into a
+        // SerializedPlan (and that takes too much time).
+        let session_state = ctx.state();
+        let execution_plan = session_state.query_planner().create_physical_plan(pre_serialized_plan.logical_plan(), &session_state).await?;
         Ok((
-            ctx.clone()
-                .state()
-                .create_physical_plan(pre_serialized_plan.logical_plan())
-                .await?,
+            execution_plan,
             pre_serialized_plan.logical_plan().clone(),
         ))
     }
@@ -346,12 +343,11 @@ impl QueryExecutor for QueryExecutorImpl {
             worker_planning_params,
             data_loaded_size,
         )?;
-        let plan_ctx = ctx.clone();
+        // We don't want to use session_state.create_physical_plan(...); see comment in router_plan.
+        let session_state = ctx.state();
+        let execution_plan = session_state.query_planner().create_physical_plan(pre_serialized_plan.logical_plan(), &session_state).await?;
         Ok((
-            plan_ctx
-                .state()
-                .create_physical_plan(pre_serialized_plan.logical_plan())
-                .await?,
+            execution_plan,
             pre_serialized_plan.logical_plan().clone(),
         ))
     }