fix(executor): fix broken merge into source profiling (#15415)

Author: zhang2014

src/query/service/src/interpreters/interpreter_merge_into.rs

@@ -94,6 +94,7 @@ impl Interpreter for MergeIntoInterpreter {
     #[async_backtrace::framed]
     async fn execute2(&self) -> Result<PipelineBuildResult> {
         let (physical_plan, _) = self.build_physical_plan().await?;
+
         let mut build_res =
             build_query_pipeline_without_render_result_set(&self.ctx, &physical_plan).await?;
 

src/query/service/src/pipelines/pipeline_builder.rs

@@ -24,6 +24,7 @@ use databend_common_pipeline_core::processors::PlanScope;
 use databend_common_pipeline_core::processors::PlanScopeGuard;
 use databend_common_pipeline_core::Pipeline;
 use databend_common_settings::Settings;
+use databend_common_sql::binder::MergeIntoType;
 use databend_common_sql::executor::PhysicalPlan;
 use databend_common_sql::IndexType;
 
@@ -98,6 +99,9 @@ impl PipelineBuilder {
     pub(crate) fn add_plan_scope(&mut self, plan: &PhysicalPlan) -> Result<Option<PlanScopeGuard>> {
         match plan {
             PhysicalPlan::EvalScalar(v) if v.exprs.is_empty() => Ok(None),
+            PhysicalPlan::MergeIntoSource(v) if v.merge_type != MergeIntoType::FullOperation => {
+                Ok(None)
+            }
             _ => {
                 let desc = plan.get_desc()?;
                 let plan_labels = plan.get_labels()?;