refactor(interpreter): move execute into trait

Author: zhang2014

src/query/service/src/interpreters/async_insert_queue.rs

@@ -506,12 +506,12 @@ impl AsyncInsertManager {
         let blocks = Arc::new(Mutex::new(VecDeque::from_iter(
             data.entries.iter().map(|x| x.block.clone()),
         )));
-        let source = BlocksSource::create(ctx, output_port.clone(), blocks)?;
+        let source = BlocksSource::create(ctx.clone(), output_port.clone(), blocks)?;
         let mut builder = SourcePipeBuilder::create();
         builder.add_source(output_port.clone(), source);
 
         interpreter.set_source_pipe_builder(Some(builder))?;
-        interpreter.execute().await?;
+        interpreter.execute(ctx).await?;
         Ok(())
     }
 

src/query/service/src/interpreters/execute_query.rs

@@ -0,0 +1,55 @@
+use std::sync::Arc;
+use common_catalog::table_context::TableContext;
+use common_pipeline_core::{Pipe, SourcePipeBuilder};
+use crate::interpreters::InterpreterFactory;
+use crate::optimizers::Optimizers;
+use crate::sessions::QueryContext;
+use crate::sql::PlanParser;
+use common_exception::{ErrorCode, Result};
+use crate::pipelines::executor::{ExecutorSettings, PipelineCompleteExecutor};
+use crate::pipelines::PipelineBuildResult;
+
+fn execute_complete(ctx: Arc<QueryContext>, build_res: PipelineBuildResult) -> Result<()> {
+    if !build_res.main_pipeline.is_complete_pipeline()? {
+        return Err(ErrorCode::LogicalError(""));
+    }
+
+    let mut pipelines = build_res.sources_pipelines;
+    pipelines.push(build_res.main_pipeline);
+
+    let settings = ctx.get_settings();
+    let query_need_abort = ctx.query_need_abort();
+    let executor_settings = ExecutorSettings::try_create(&settings)?;
+    PipelineCompleteExecutor::from_pipelines(query_need_abort, pipelines, executor_settings)?
+        .execute()
+}
+
+pub async fn execute_in(ctx: Arc<QueryContext>, query: &str, pipe: Pipe) -> Result<()> {
+    match pipe {
+        Pipe::ResizePipe { .. } => Err(ErrorCode::LogicalError("")),
+        Pipe::SimplePipe { processors, inputs_port, outputs_port } => {
+            let plan = PlanParser::parse(ctx.clone(), query).await?;
+            let optimized = Optimizers::create(ctx.clone()).optimize(&plan)?;
+            let interpreter = InterpreterFactory::get(ctx.clone(), optimized)?;
+
+            let mut source_pipe_builder = SourcePipeBuilder::create();
+            for (index, processor) in processors.into_iter().enumerate() {
+                source_pipe_builder.add_source(outputs_port[index].clone(), processor);
+            }
+
+            interpreter.set_source_pipe_builder(Some(source_pipe_builder))?;
+            execute_complete(ctx, interpreter.execute2().await?)
+        }
+    }
+}
+
+pub async fn execute(ctx: Arc<QueryContext>, query: &str) -> Result<()> {
+    let plan = PlanParser::parse(ctx.clone(), query).await?;
+    let optimized = Optimizers::create(ctx.clone()).optimize(&plan)?;
+    let interpreter = InterpreterFactory::get(ctx.clone(), optimized)?;
+
+    match interpreter.execute2().await?.main_pipeline.pipes.is_empty() {
+        true => Ok(()),
+        false => Err(ErrorCode::LogicalError(""))
+    }
+}

src/query/service/src/interpreters/interpreter.rs

@@ -12,14 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use common_datavalues::DataSchemaRef;
+use std::sync::Arc;
+use common_catalog::table_context::TableContext;
+use common_datavalues::{DataSchema, DataSchemaRef};
 use common_datavalues::DataSchemaRefExt;
 use common_exception::ErrorCode;
 use common_exception::Result;
-use common_streams::SendableDataBlockStream;
+use common_streams::{DataBlockStream, SendableDataBlockStream};
+use crate::interpreters::ProcessorExecutorStream;
+use crate::pipelines::executor::{ExecutorSettings, PipelineCompleteExecutor, PipelinePullingExecutor};
 
 use crate::pipelines::PipelineBuildResult;
 use crate::pipelines::SourcePipeBuilder;
+use crate::sessions::QueryContext;
 
 #[async_trait::async_trait]
 /// Interpreter is a trait for different PlanNode
@@ -34,8 +39,38 @@ pub trait Interpreter: Sync + Send {
     }
 
     /// The core of the databend processor which will execute the logical plan and get the DataBlock
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
-        unimplemented!()
+    async fn execute(&self, ctx: Arc<QueryContext>) -> Result<SendableDataBlockStream> {
+        let build_res = self.execute2().await?;
+
+        let settings = ctx.get_settings();
+        let query_need_abort = ctx.query_need_abort();
+        let executor_settings = ExecutorSettings::try_create(&settings)?;
+
+        if build_res.main_pipeline.is_complete_pipeline()? {
+            let mut pipelines = build_res.sources_pipelines;
+            pipelines.push(build_res.main_pipeline);
+
+            let complete_executor = PipelineCompleteExecutor::from_pipelines(
+                query_need_abort,
+                pipelines,
+                executor_settings,
+            )?;
+
+            complete_executor.execute()?;
+            return Ok(Box::pin(DataBlockStream::create(
+                Arc::new(DataSchema::new(vec![])),
+                None,
+                vec![],
+            )));
+        }
+
+        Ok(Box::pin(Box::pin(ProcessorExecutorStream::create(
+            PipelinePullingExecutor::from_pipelines(
+                ctx.query_need_abort(),
+                build_res,
+                executor_settings,
+            )?,
+        )?)))
     }
 
     /// The core of the databend processor which will execute the logical plan and build the pipeline

src/query/service/src/interpreters/interpreter_copy_v2.rs

@@ -275,30 +275,6 @@ impl Interpreter for CopyInterpreterV2 {
     }
 
     #[tracing::instrument(level = "debug", name = "copy_interpreter_execute_v2", skip(self), fields(ctx.id = self.ctx.get_id().as_str()))]
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
-        let build_res = self.execute2().await?;
-
-        let settings = self.ctx.get_settings();
-        let query_need_abort = self.ctx.query_need_abort();
-        let executor_settings = ExecutorSettings::try_create(&settings)?;
-
-        let mut pipelines = build_res.sources_pipelines;
-        pipelines.push(build_res.main_pipeline);
-
-        let executor = PipelineCompleteExecutor::from_pipelines(
-            query_need_abort,
-            pipelines,
-            executor_settings,
-        )?;
-
-        executor.execute()?;
-        Ok(Box::pin(DataBlockStream::create(
-            Arc::new(DataSchema::new(vec![])),
-            None,
-            vec![],
-        )))
-    }
-
     async fn execute2(&self) -> Result<PipelineBuildResult> {
         match &self.plan {
             // TODO(xuanwo): extract them as a separate function.

src/query/service/src/interpreters/interpreter_factory_interceptor.rs

@@ -73,7 +73,7 @@ impl Interpreter for InterceptorInterpreter {
         self.inner.schema()
     }
 
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
+    async fn execute(&self, ctx: Arc<QueryContext>) -> Result<SendableDataBlockStream> {
         // Management mode access check.
         match &self.new_plan {
             Some(p) => self.management_mode_access.check_new(p)?,
@@ -83,7 +83,7 @@ impl Interpreter for InterceptorInterpreter {
         let _ = self
             .inner
             .set_source_pipe_builder((*self.source_pipe_builder.lock()).clone());
-        let result_stream = match self.inner.execute().await {
+        let result_stream = match self.inner.execute(ctx).await {
             Ok(s) => s,
             Err(e) => {
                 self.ctx.set_error(e.clone());

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

@@ -85,13 +85,6 @@ impl Interpreter for InsertInterpreter {
         "InsertIntoInterpreter"
     }
 
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
-        let build_res = self.execute2().await?;
-
-        execute_pipeline(self.ctx.clone(), build_res)?;
-        Ok(Box::pin(DataBlockStream::create(self.plan.schema(), None, vec![])))
-    }
-
     async fn execute2(&self) -> Result<PipelineBuildResult> {
         let plan = &self.plan;
         let settings = self.ctx.get_settings();

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

@@ -83,14 +83,6 @@ impl Interpreter for InsertInterpreterV2 {
         "InsertIntoInterpreter"
     }
 
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
-        let build_res = self.execute2().await?;
-        execute_pipeline(self.ctx.clone(), build_res)?;
-
-        // Ok(PipelineBuildResult::create())
-        unimplemented!()
-    }
-
     async fn execute2(&self) -> Result<PipelineBuildResult> {
         let plan = &self.plan;
         let settings = self.ctx.get_settings();

src/query/service/src/interpreters/interpreter_select.rs

@@ -51,23 +51,6 @@ impl SelectInterpreter {
             &self.select.input,
         )
     }
-
-    async fn build_pipeline(&self) -> Result<PipelineBuildResult> {
-        match self.ctx.get_cluster().is_empty() {
-            true => {
-                let settings = self.ctx.get_settings();
-                let builder = QueryPipelineBuilder::create(self.ctx.clone());
-                let mut query_pipeline = builder.finalize(&self.rewrite_plan()?)?;
-                query_pipeline.set_max_threads(settings.get_max_threads()? as usize);
-                Ok(query_pipeline)
-            }
-            false => {
-                let ctx = self.ctx.clone();
-                let optimized_plan = self.rewrite_plan()?;
-                plan_schedulers::schedule_query_new(ctx, &optimized_plan).await
-            }
-        }
-    }
 }
 
 #[async_trait::async_trait]
@@ -86,22 +69,20 @@ impl Interpreter for SelectInterpreter {
     /// Currently, the method has two sets of logic, if `get_enable_new_processor_framework` is turned on in the settings,
     /// the execution will use the new processor, otherwise the old processing logic will be executed.
     /// Note: there is an issue to track the progress of the new processor:  https://github.com/datafuselabs/databend/issues/3379
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
-        let build_res = self.build_pipeline().await?;
-        let query_need_abort = self.ctx.query_need_abort();
-        let executor_settings = ExecutorSettings::try_create(&self.ctx.get_settings())?;
-        Ok(Box::pin(ProcessorExecutorStream::create(
-            PipelinePullingExecutor::from_pipelines(
-                query_need_abort,
-                build_res,
-                executor_settings,
-            )?,
-        )?))
-    }
-
-    /// This method will create a new pipeline
-    /// The QueryPipelineBuilder will use the optimized plan to generate a Pipeline
     async fn execute2(&self) -> Result<PipelineBuildResult> {
-        self.build_pipeline().await
+        match self.ctx.get_cluster().is_empty() {
+            true => {
+                let settings = self.ctx.get_settings();
+                let builder = QueryPipelineBuilder::create(self.ctx.clone());
+                let mut query_pipeline = builder.finalize(&self.rewrite_plan()?)?;
+                query_pipeline.set_max_threads(settings.get_max_threads()? as usize);
+                Ok(query_pipeline)
+            }
+            false => {
+                let ctx = self.ctx.clone();
+                let optimized_plan = self.rewrite_plan()?;
+                plan_schedulers::schedule_query_new(ctx, &optimized_plan).await
+            }
+        }
     }
 }

src/query/service/src/interpreters/interpreter_select_v2.rs

@@ -91,7 +91,7 @@ impl Interpreter for SelectInterpreterV2 {
     }
 
     #[tracing::instrument(level = "debug", name = "select_interpreter_v2_execute", skip(self), fields(ctx.id = self.ctx.get_id().as_str()))]
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
+    async fn execute(&self, ctx: Arc<QueryContext>) -> Result<SendableDataBlockStream> {
         let build_res = self.build_pipeline().await?;
 
         // WTF: We need to implement different logic for the HTTP handler

src/query/service/src/interpreters/interpreter_table_create_v2.rs

@@ -160,7 +160,7 @@ impl CreateTableInterpreterV2 {
         };
         let insert_interpreter_v2 =
             InsertInterpreterV2::try_create(self.ctx.clone(), insert_plan, false)?;
-        insert_interpreter_v2.execute().await?;
+        insert_interpreter_v2.execute(self.ctx.clone()).await?;
 
         Ok(PipelineBuildResult::create())
     }