fix(query): align profiles with physical plan scope

Author: dqhl76

src/common/base/src/runtime/profile/profile.rs

@@ -47,6 +47,7 @@ pub struct Profile {
     pub plan_id: Option<u32>,
     pub plan_name: Option<String>,
     pub plan_parent_id: Option<u32>,
+    pub plan_group_id: u64,
     pub labels: Arc<Vec<ProfileLabel>>,
     pub title: Arc<String>,
 
@@ -65,6 +66,7 @@ impl Clone for Profile {
             plan_id: self.plan_id,
             plan_name: self.plan_name.clone(),
             plan_parent_id: self.plan_parent_id,
+            plan_group_id: self.plan_group_id,
             labels: self.labels.clone(),
             title: self.title.clone(),
             metrics_registry: self.metrics_registry.clone(),
@@ -87,6 +89,7 @@ impl Profile {
         plan_id: Option<u32>,
         plan_name: Option<String>,
         plan_parent_id: Option<u32>,
+        plan_group_id: u64,
         title: Arc<String>,
         labels: Arc<Vec<ProfileLabel>>,
         metrics_registry: Option<Arc<ScopedRegistry>>,
@@ -97,6 +100,7 @@ impl Profile {
             plan_id,
             plan_name,
             plan_parent_id,
+            plan_group_id,
             title,
             labels,
             statistics: Self::create_items(),

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

@@ -94,6 +94,8 @@ pub struct PlanProfile {
     pub id: Option<u32>,
     pub name: Option<String>,
     pub parent_id: Option<u32>,
+    #[serde(default)]
+    pub group_id: u64,
     pub title: Arc<String>,
     pub labels: Arc<Vec<ProfileLabel>>,
 
@@ -136,6 +138,7 @@ impl PlanProfile {
             id: profile.plan_id,
             name: profile.plan_name.clone(),
             parent_id: profile.plan_parent_id,
+            group_id: profile.plan_group_id,
             title: profile.title.clone(),
             labels: profile.labels.clone(),
             metrics: BTreeMap::new(),
@@ -198,6 +201,8 @@ pub struct PlanScope {
     pub id: u32,
     pub name: String,
     pub parent_id: Option<u32>,
+    /// The execution unit (physical plan tree) this scope belongs to. See `Profile::plan_group_id`.
+    pub group_id: u64,
     pub title: Arc<String>,
     pub labels: Arc<Vec<ProfileLabel>>,
     pub metrics_registry: Arc<ScopedRegistry>,
@@ -211,6 +216,7 @@ impl PlanScope {
     pub fn create(
         id: u32,
         name: String,
+        group_id: u64,
         title: Arc<String>,
         labels: Arc<Vec<ProfileLabel>>,
     ) -> Arc<PlanScope> {
@@ -220,6 +226,7 @@ impl PlanScope {
             labels,
             title,
             parent_id,
+            group_id,
             name,
             metrics_registry: ScopedRegistry::create(None),
         })

src/query/service/src/interpreters/interpreter_copy_into_location.rs

@@ -105,6 +105,7 @@ impl CopyIntoLocationInterpreter {
 
         let mut next_plan_id = 0;
         physical_plan.adjust_plan_id(&mut next_plan_id);
+        physical_plan.assign_profile_group_id();
         Ok((
             build_query_pipeline_without_render_result_set(&self.ctx, &physical_plan).await?,
             update_stream_meta_req,

src/query/service/src/interpreters/interpreter_copy_into_table.rs

@@ -333,6 +333,7 @@ impl CopyIntoTableInterpreter {
 
         let mut next_plan_id = 0;
         root.adjust_plan_id(&mut next_plan_id);
+        root.assign_profile_group_id();
 
         Ok((root, update_stream_meta_reqs, new_schema))
     }

src/query/service/src/interpreters/interpreter_explain.rs

@@ -67,7 +67,6 @@ use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 use crate::sessions::TableContextPartitionStats;
 use crate::sessions::TableContextQueryIdentity;
-use crate::sessions::TableContextQueryProfile;
 use crate::sessions::TableContextRuntimeFilter;
 use crate::sessions::TableContextSettings;
 use crate::sql::optimizer::ir::SExpr;
@@ -460,10 +459,11 @@ impl ExplainInterpreter {
 
         let mut builder = PhysicalPlanBuilder::new(metadata.clone(), self.ctx.clone(), true);
         let plan = builder.build(s_expr, required).await?;
+        let main_group_id = plan.get_meta().profile_group_id;
         let build_res = build_query_pipeline(&self.ctx, &[], &plan, ignore_result).await?;
 
         // Drain the data
-        let query_profiles = self.execute_and_get_profiles(build_res).await?;
+        let query_profiles = self.execute_and_get_profiles(build_res, main_group_id).await?;
 
         Ok(GraphicalProfiles {
             query_id: query_ctx.get_id(),
@@ -485,10 +485,11 @@ impl ExplainInterpreter {
             builder.set_mutation_build_info(build_info);
         }
         let mut plan = builder.build(s_expr, required).await?;
+        let main_group_id = plan.get_meta().profile_group_id;
         let build_res = build_query_pipeline(&self.ctx, &[], &plan, ignore_result).await?;
 
         // Drain the data
-        let query_profiles = self.execute_and_get_profiles(build_res).await?;
+        let query_profiles = self.execute_and_get_profiles(build_res, main_group_id).await?;
 
         let mut pruned_partitions_stats = self.ctx.get_pruned_partitions_stats();
         if !pruned_partitions_stats.is_empty() {
@@ -541,6 +542,7 @@ impl ExplainInterpreter {
     async fn execute_and_get_profiles(
         &self,
         mut build_res: PipelineBuildResult,
+        main_group_id: u64,
     ) -> Result<HashMap<u32, PlanProfile>> {
         let settings = self.ctx.get_settings();
         build_res.set_max_threads(settings.get_max_threads()? as usize);
@@ -563,9 +565,12 @@ impl ExplainInterpreter {
                 while (executor.pull_data().await?).is_some() {}
             }
         }
+        // EXPLAIN ANALYZE renders the main query plan tree only and looks up each node by its
+        // original plan id. Restrict to the main plan's profile group so that profiles from other
+        // execution units (e.g. materialized CTE fills) with overlapping plan ids are not matched.
         Ok(self
             .ctx
-            .get_query_profiles()
+            .get_query_profiles_by_group(main_group_id)
             .into_iter()
             .filter(|x| x.id.is_some())
             .map(|x| (x.id.unwrap(), x))

src/query/service/src/interpreters/interpreter_insert.rs

@@ -259,6 +259,7 @@ impl Interpreter for InsertInterpreter {
                 };
 
                 insert_select_plan.adjust_plan_id(&mut 0);
+                insert_select_plan.assign_profile_group_id();
                 let mut build_res =
                     build_query_pipeline_without_render_result_set(&self.ctx, &insert_select_plan)
                         .await?;

src/query/service/src/interpreters/interpreter_insert_multi_table.rs

@@ -243,6 +243,7 @@ impl InsertMultiTableInterpreter {
 
         let mut next_plan_id = 0;
         root.adjust_plan_id(&mut next_plan_id);
+        root.assign_profile_group_id();
         Ok(root)
     }
 

src/query/service/src/interpreters/interpreter_replace.rs

@@ -378,6 +378,7 @@ impl ReplaceInterpreter {
         });
 
         root.adjust_plan_id(&mut 0);
+        root.assign_profile_group_id();
         Ok((root, purge_info))
     }
 

src/query/service/src/interpreters/interpreter_table_modify_column.rs

@@ -912,6 +912,7 @@ pub(crate) async fn build_select_insert_plan(
 
     let mut index = 0;
     insert_plan.adjust_plan_id(&mut index);
+    insert_plan.assign_profile_group_id();
     let mut build_res = build_query_pipeline_without_render_result_set(&ctx, &insert_plan).await?;
 
     // 5. commit new meta schema and snapshots

src/query/service/src/interpreters/interpreter_table_recluster.rs

@@ -259,6 +259,7 @@ impl ReclusterTableInterpreter {
             return Ok(true);
         };
         physical_plan.adjust_plan_id(&mut 0);
+        physical_plan.assign_profile_group_id();
         let mut build_res =
             build_query_pipeline_without_render_result_set(&self.ctx, &physical_plan).await?;
         {