refactor(query): align profiles with physical plan scope

Author: dqhl76

src/query/catalog/src/table_context/query.rs

@@ -68,7 +68,7 @@ pub trait TableContextQueryInfo: Send + Sync {
 pub trait TableContextQueryProfile: Send + Sync {
     fn get_queries_profile(&self) -> HashMap<String, Vec<PlanProfile>>;
 
-    fn add_query_profiles(&self, profiles: &HashMap<u32, PlanProfile>);
+    fn add_query_profiles(&self, profile_scope_id: &str, profiles: &HashMap<u32, PlanProfile>);
 
     fn get_query_profiles(&self) -> Vec<PlanProfile>;
 }

src/query/ee/src/storages/fuse/operations/virtual_columns.rs

@@ -702,7 +702,8 @@ async fn execute_complete_pipeline(
     if build_res.main_pipeline.is_complete_pipeline()? {
         let mut pipelines = build_res.sources_pipelines;
         pipelines.push(build_res.main_pipeline);
-        let executor_settings = ExecutorSettings::try_create(ctx)?;
+        let executor_settings =
+            ExecutorSettings::try_create(ctx, build_res.profile_scope_id.clone())?;
         let complete_executor =
             PipelineCompleteExecutor::from_pipelines(pipelines, executor_settings)?;
         complete_executor.execute().await?;

src/query/ee/tests/it/storages/fuse/operations/virtual_columns.rs

@@ -65,7 +65,7 @@ async fn setup_table_with_virtual_columns(num_blocks: usize) -> anyhow::Result<T
         .await?;
         let settings = ctx.get_settings();
         build_res.set_max_threads(settings.get_max_threads()? as usize);
-        let settings = ExecutorSettings::try_create(ctx.clone())?;
+        let settings = ExecutorSettings::try_create_with_new_profile_scope(ctx.clone())?;
         if build_res.main_pipeline.is_complete_pipeline()? {
             let mut pipelines = build_res.sources_pipelines;
             pipelines.push(build_res.main_pipeline);
@@ -131,7 +131,7 @@ async fn test_fuse_do_refresh_virtual_column() -> anyhow::Result<()> {
 
     let settings = table_ctx.get_settings();
     build_res.set_max_threads(settings.get_max_threads()? as usize);
-    let settings = ExecutorSettings::try_create(table_ctx.clone())?;
+    let settings = ExecutorSettings::try_create_with_new_profile_scope(table_ctx.clone())?;
 
     if build_res.main_pipeline.is_complete_pipeline()? {
         let mut pipelines = build_res.sources_pipelines;

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

@@ -37,11 +37,22 @@ pub enum CallbackType {
 pub struct ExecutionInfo {
     pub res: Result<()>,
     pub profiling: HashMap<u32, PlanProfile>,
+    // Profiles produced by implementation-only child executors can share the same scope
+    // as their parent physical plan.
+    pub profile_scope_id: String,
 }
 
 impl ExecutionInfo {
-    pub fn create(res: Result<()>, profiling: HashMap<u32, PlanProfile>) -> ExecutionInfo {
-        ExecutionInfo { res, profiling }
+    pub fn create(
+        res: Result<()>,
+        profiling: HashMap<u32, PlanProfile>,
+        profile_scope_id: impl Into<String>,
+    ) -> ExecutionInfo {
+        ExecutionInfo {
+            res,
+            profiling,
+            profile_scope_id: profile_scope_id.into(),
+        }
     }
 }
 
@@ -314,7 +325,7 @@ mod tests {
             );
         }
 
-        chain.apply(ExecutionInfo::create(Ok(()), HashMap::new()))?;
+        chain.apply(ExecutionInfo::create(Ok(()), HashMap::new(), String::new()))?;
 
         assert_eq!(seq.load(Ordering::SeqCst), 10);
 
@@ -378,7 +389,7 @@ mod tests {
             }),
         );
 
-        chain.apply(ExecutionInfo::create(Ok(()), HashMap::new()))?;
+        chain.apply(ExecutionInfo::create(Ok(()), HashMap::new(), String::new()))?;
 
         assert_eq!(seq.load(Ordering::SeqCst), 13);
 
@@ -443,7 +454,7 @@ mod tests {
             }),
         );
 
-        chain.apply(ExecutionInfo::create(Ok(()), HashMap::new()))?;
+        chain.apply(ExecutionInfo::create(Ok(()), HashMap::new(), String::new()))?;
 
         assert_eq!(seq.load(Ordering::SeqCst), 13);
 
@@ -522,7 +533,7 @@ mod tests {
 
         assert!(
             chain
-                .apply(ExecutionInfo::create(Ok(()), HashMap::new()))
+                .apply(ExecutionInfo::create(Ok(()), HashMap::new(), String::new(),))
                 .is_err()
         );
 

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

@@ -646,9 +646,11 @@ impl Drop for Pipeline {
                 "Pipeline illegal state: not successfully shutdown.",
             ));
 
-            let _ = self
-                .on_finished_chain
-                .apply(ExecutionInfo::create(cause, HashMap::new()));
+            let _ = self.on_finished_chain.apply(ExecutionInfo::create(
+                cause,
+                HashMap::new(),
+                String::new(),
+            ));
         })
     }
 }

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

@@ -213,8 +213,9 @@ impl PlanScope {
         name: String,
         title: Arc<String>,
         labels: Arc<Vec<ProfileLabel>>,
+        parent_id: Option<u32>,
     ) -> Arc<PlanScope> {
-        let parent_id = PlanScope::get_plan_scope().map(|x| x.id);
+        let parent_id = parent_id.or_else(|| PlanScope::get_plan_scope().map(|x| x.id));
         Arc::new(PlanScope {
             id,
             labels,

src/query/service/src/interpreters/common/finish_hook.rs

@@ -93,7 +93,7 @@ impl QueryFinishHooks {
                 ctx.collect_local_perf_counters(node_id);
             }
             if self.collect_profiles {
-                ctx.add_query_profiles(&info.profiling);
+                ctx.add_query_profiles(&info.profile_scope_id, &info.profiling);
             }
             let hooks_res = if self.run_hooks {
                 run_hooks(ctx.clone())

src/query/service/src/interpreters/hook/analyze_hook.rs

@@ -89,7 +89,7 @@ async fn do_analyze(ctx: Arc<QueryContext>, desc: AnalyzeDesc) -> Result<()> {
         false,
     )?;
     pipeline.set_max_threads(ctx.get_settings().get_max_threads()? as usize);
-    let executor_settings = ExecutorSettings::try_create(ctx.clone())?;
+    let executor_settings = ExecutorSettings::try_create_with_new_profile_scope(ctx.clone())?;
     let pipelines = vec![pipeline];
     let complete_executor = PipelineCompleteExecutor::from_pipelines(pipelines, executor_settings)?;
     ctx.set_executor(complete_executor.get_inner())?;

src/query/service/src/interpreters/hook/compact_hook.rs

@@ -199,7 +199,8 @@ async fn compact_table(
             ));
 
             build_res.set_max_threads(settings.get_max_threads()? as usize);
-            let executor_settings = ExecutorSettings::try_create(ctx.clone())?;
+            let executor_settings =
+                ExecutorSettings::try_create(ctx.clone(), build_res.profile_scope_id.clone())?;
 
             let mut pipelines = build_res.sources_pipelines;
             pipelines.push(build_res.main_pipeline);

src/query/service/src/interpreters/hook/refresh_hook.rs

@@ -125,7 +125,10 @@ async fn do_refresh(ctx: Arc<QueryContext>, desc: RefreshDesc) -> Result<()> {
 
                     let settings = ctx_cloned.get_settings();
                     build_res.set_max_threads(settings.get_max_threads()? as usize);
-                    let settings = ExecutorSettings::try_create(ctx_cloned.clone())?;
+                    let settings = ExecutorSettings::try_create(
+                        ctx_cloned.clone(),
+                        build_res.profile_scope_id.clone(),
+                    )?;
 
                     if build_res.main_pipeline.is_complete_pipeline()? {
                         let query_ctx = ctx_cloned.clone();
@@ -162,7 +165,10 @@ async fn do_refresh(ctx: Arc<QueryContext>, desc: RefreshDesc) -> Result<()> {
 
                     let settings = ctx_cloned.get_settings();
                     build_res.set_max_threads(settings.get_max_threads()? as usize);
-                    let settings = ExecutorSettings::try_create(ctx_cloned.clone())?;
+                    let settings = ExecutorSettings::try_create(
+                        ctx_cloned.clone(),
+                        build_res.profile_scope_id.clone(),
+                    )?;
 
                     if build_res.main_pipeline.is_complete_pipeline()? {
                         let query_ctx = ctx_cloned.clone();