fix(processor): try fix test failure

Author: zhang2014

src/query/catalog/src/table_context.rs

@@ -108,4 +108,6 @@ pub trait TableContext: Send + Sync {
     -> Result<Arc<dyn Table>>;
     fn get_processes_info(&self) -> Vec<ProcessInfo>;
     fn get_runtime(&self) -> Result<Arc<Runtime>>;
+
+    fn clone_inner(&self) -> Arc<dyn TableContext>;
 }

src/query/service/src/api/rpc/exchange/exchange_manager.rs

@@ -668,7 +668,6 @@ impl FragmentCoordinator {
 
             match &self.payload {
                 FragmentPayload::PlanV2(plan) => {
-                    let ctx = QueryContext::create_from(ctx);
                     let pipeline_builder = PipelineBuilderV2::create(ctx);
                     self.pipeline_build_res = Some(pipeline_builder.finalize(plan)?);
                 }

src/query/service/src/interpreters/interpreter_insert_v2.rs

@@ -164,7 +164,6 @@ impl Interpreter for InsertInterpreterV2 {
                         _ => unreachable!(),
                     };
 
-                    table1.get_table_info();
                     let catalog = self.plan.catalog.clone();
                     let is_distributed_plan = select_plan.is_distributed_plan();
 

src/query/service/src/sessions/query_ctx.rs

@@ -386,6 +386,10 @@ impl TableContext for QueryContext {
     fn get_runtime(&self) -> Result<Arc<Runtime>> {
         self.shared.try_get_runtime()
     }
+
+    fn clone_inner(&self) -> Arc<dyn TableContext> {
+        QueryContext::create_from_shared(self.shared.clone())
+    }
 }
 
 impl TrySpawn for QueryContext {

src/query/storages/fuse/src/operations/read.rs

@@ -78,6 +78,7 @@ impl FuseTable {
         plan: &ReadDataSourcePlan,
         pipeline: &mut Pipeline,
     ) -> Result<()> {
+        let ctx = ctx.clone_inner();
         let mut lazy_init_segments = Vec::with_capacity(plan.parts.len());
 
         for part in &plan.parts {