fix(executor): fix broken profile graph for union all

src/query/pipeline/core/src/pipeline.rs

@@ -62,7 +62,7 @@ pub struct Pipeline {
     on_finished: Option<FinishedCallback>,
     lock_guards: Vec<LockGuard>,
 
-    pub plans_scope: Vec<PlanScope>,
+    plans_scope: Vec<PlanScope>,
     scope_size: Arc<AtomicUsize>,
 }
 
@@ -154,6 +154,11 @@ impl Pipeline {
         self
     }
 
+    pub fn get_scopes(&self) -> Vec<PlanScope> {
+        let scope_size = self.scope_size.load(Ordering::SeqCst);
+        self.plans_scope[..scope_size].to_vec()
+    }
+
     pub fn add_pipe(&mut self, mut pipe: Pipe) {
         let (scope_idx, _) = self.scope_size.load(Ordering::SeqCst).overflowing_sub(1);
 
@@ -454,7 +459,7 @@ impl Pipeline {
 
         if self.plans_scope.len() > scope_idx {
             self.plans_scope[scope_idx] = scope;
-            self.plans_scope.shrink_to(scope_idx + 1);
+            self.plans_scope.truncate(scope_idx + 1);
             return PlanScopeGuard::create(self.scope_size.clone(), scope_idx);
         }
 

src/query/service/src/pipelines/builders/builder_join.rs

@@ -99,7 +99,7 @@ impl PipelineBuilder {
             right_side_context,
             self.enable_profiling,
             self.proc_profs.clone(),
-            self.main_pipeline.plans_scope.clone(),
+            self.main_pipeline.get_scopes(),
         );
         right_side_builder.cte_state = self.cte_state.clone();
         let mut right_res = right_side_builder.finalize(&range_join.right)?;
@@ -154,7 +154,7 @@ impl PipelineBuilder {
             build_side_context,
             self.enable_profiling,
             self.proc_profs.clone(),
-            self.main_pipeline.plans_scope.clone(),
+            self.main_pipeline.get_scopes(),
         );
         build_side_builder.cte_state = self.cte_state.clone();
         let mut build_res = build_side_builder.finalize(build)?;
@@ -317,7 +317,7 @@ impl PipelineBuilder {
             left_side_ctx,
             self.enable_profiling,
             self.proc_profs.clone(),
-            self.main_pipeline.plans_scope.clone(),
+            self.main_pipeline.get_scopes(),
         );
         left_side_builder.cte_state = self.cte_state.clone();
         let mut left_side_pipeline = left_side_builder.finalize(left_side)?;

src/query/service/src/pipelines/builders/builder_union_all.rs

@@ -65,9 +65,10 @@ impl PipelineBuilder {
             union_ctx,
             self.enable_profiling,
             self.proc_profs.clone(),
-            self.main_pipeline.plans_scope.clone(),
+            self.main_pipeline.get_scopes(),
         );
         pipeline_builder.cte_state = self.cte_state.clone();
+
         let mut build_res = pipeline_builder.finalize(input)?;
 
         assert!(build_res.main_pipeline.is_pulling_pipeline()?);