refactor(interpreter): move execute to execute2

Author: zhang2014

src/query/pipeline/sources/src/processors/sources/mod.rs

@@ -22,6 +22,7 @@ pub mod stream_source;
 pub mod stream_source_v2;
 pub mod sync_source;
 pub mod sync_source_receiver;
+mod one_block_source;
 
 pub use async_source::AsyncSource;
 pub use async_source::AsyncSourcer;
@@ -40,6 +41,7 @@ pub use sync_source::SyncSource;
 pub use sync_source::SyncSourcer;
 pub use sync_source::*;
 pub use sync_source_receiver::SyncReceiverSource;
+pub use one_block_source::OneBlockSource;
 
 #[allow(dead_code)]
 mod source_example {

src/query/pipeline/sources/src/processors/sources/one_block_source.rs

@@ -0,0 +1,71 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::any::Any;
+use std::collections::VecDeque;
+use std::sync::Arc;
+
+use common_catalog::table_context::TableContext;
+use common_datablocks::DataBlock;
+use common_exception::Result;
+use common_pipeline_core::processors::port::OutputPort;
+use common_pipeline_core::processors::processor::{Event, ProcessorPtr};
+use parking_lot::Mutex;
+use common_pipeline_core::processors::Processor;
+
+use crate::processors::sources::SyncSource;
+use crate::processors::sources::SyncSourcer;
+
+pub struct OneBlockSource {
+    output: Arc<OutputPort>,
+    data_block: Option<DataBlock>,
+}
+
+impl OneBlockSource {
+    pub fn create(output: Arc<OutputPort>, data_block: DataBlock) -> Result<ProcessorPtr> {
+        Ok(ProcessorPtr::create(Box::new(OneBlockSource {
+            output,
+            data_block: Some(data_block),
+        })))
+    }
+}
+
+#[async_trait::async_trait]
+impl Processor for OneBlockSource {
+    fn name(&self) -> &'static str {
+        "BlockSource"
+    }
+
+    fn as_any(&mut self) -> &mut dyn Any {
+        self
+    }
+
+    fn event(&mut self) -> Result<Event> {
+        if self.output.is_finished() {
+            return Ok(Event::Finished);
+        }
+
+        if !self.output.can_push() {
+            return Ok(Event::NeedConsume);
+        }
+
+        if let Some(data_block) = self.data_block.take() {
+            self.output.push_data(Ok(data_block));
+            return Ok(Event::NeedConsume);
+        }
+
+        self.output.finish();
+        Ok(Event::Finished)
+    }
+}

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

@@ -393,7 +393,7 @@ impl AsyncInsertManager {
                     input: Arc::new((**plan).clone()),
                 })?;
 
-                let mut build_res = select_interpreter.create_new_pipeline().await?;
+                let mut build_res = select_interpreter.execute2().await?;
 
                 let mut sink_pipeline_builder = SinkPipeBuilder::create();
                 for _ in 0..build_res.main_pipeline.output_len() {

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

@@ -34,18 +34,13 @@ 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>;
-
-    /// Create the new pipeline for databend's new execution model
-    /// Currently databend is developing a new execution model with a hybrid pull-based & push-based strategy
-    /// The method now only is implemented by SelectInterpreter
-    async fn create_new_pipeline(&self) -> Result<PipelineBuildResult> {
-        Err(ErrorCode::UnImplement(format!(
-            "UnImplement create_new_pipeline method for {:?}",
-            self.name()
-        )))
+    async fn execute(&self) -> Result<SendableDataBlockStream> {
+        unimplemented!()
     }
 
+    /// The core of the databend processor which will execute the logical plan and build the pipeline
+    async fn execute2(&self) -> Result<PipelineBuildResult>;
+
     /// Do some start work for the interpreter
     /// Such as query counts, query start time and etc
     async fn start(&self) -> Result<()> {

src/query/service/src/interpreters/interpreter_call.rs

@@ -25,6 +25,7 @@ use super::Interpreter;
 use crate::interpreters::ProcessorExecutorStream;
 use crate::pipelines::executor::ExecutorSettings;
 use crate::pipelines::executor::PipelinePullingExecutor;
+use crate::pipelines::PipelineBuildResult;
 use crate::procedures::ProcedureFactory;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
@@ -61,7 +62,7 @@ impl Interpreter for CallInterpreter {
     }
 
     #[tracing::instrument(level = "debug", name = "call_interpreter_execute", skip(self), fields(ctx.id = self.ctx.get_id().as_str()))]
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
         let plan = &self.plan;
 
         let name = plan.name.clone();
@@ -72,18 +73,10 @@ impl Interpreter for CallInterpreter {
             *schema = Some(last_schema);
         }
 
-        let mut pipeline = Pipeline::create();
-        func.eval(self.ctx.clone(), plan.args.clone(), &mut pipeline)
+        let mut build_res = PipelineBuildResult::create();
+        func.eval(self.ctx.clone(), plan.args.clone(), &mut build_res.main_pipeline)
             .await?;
 
-        let ctx = &self.ctx;
-        let settings = ctx.get_settings();
-        let query_need_abort = ctx.query_need_abort();
-        pipeline.set_max_threads(settings.get_max_threads()? as usize);
-        let executor_settings = ExecutorSettings::try_create(&settings)?;
-        let executor =
-            PipelinePullingExecutor::try_create(query_need_abort, pipeline, executor_settings)?;
-
-        Ok(Box::pin(ProcessorExecutorStream::create(executor)?))
+        Ok(build_res)
     }
 }

src/query/service/src/interpreters/interpreter_cluster_key_alter.rs

@@ -20,6 +20,7 @@ use common_meta_types::UserPrivilegeType;
 use common_planners::AlterTableClusterKeyPlan;
 use common_streams::DataBlockStream;
 use common_streams::SendableDataBlockStream;
+use crate::pipelines::PipelineBuildResult;
 
 use super::Interpreter;
 use crate::sessions::QueryContext;
@@ -42,7 +43,7 @@ impl Interpreter for AlterTableClusterKeyInterpreter {
         "AlterTableClusterKeyInterpreter"
     }
 
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
         let plan = &self.plan;
         self.ctx
             .get_current_session()
@@ -68,10 +69,7 @@ impl Interpreter for AlterTableClusterKeyInterpreter {
         table
             .alter_table_cluster_keys(self.ctx.clone(), &self.plan.catalog, cluster_key_str)
             .await?;
-        Ok(Box::pin(DataBlockStream::create(
-            self.plan.schema(),
-            None,
-            vec![],
-        )))
+
+        Ok(PipelineBuildResult::create())
     }
 }

src/query/service/src/interpreters/interpreter_cluster_key_drop.rs

@@ -20,6 +20,7 @@ use common_meta_types::UserPrivilegeType;
 use common_planners::DropTableClusterKeyPlan;
 use common_streams::DataBlockStream;
 use common_streams::SendableDataBlockStream;
+use crate::pipelines::PipelineBuildResult;
 
 use super::Interpreter;
 use crate::sessions::QueryContext;
@@ -42,7 +43,7 @@ impl Interpreter for DropTableClusterKeyInterpreter {
         "DropTableClusterKeyInterpreter"
     }
 
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
         let plan = &self.plan;
         self.ctx
             .get_current_session()
@@ -66,10 +67,7 @@ impl Interpreter for DropTableClusterKeyInterpreter {
         table
             .drop_table_cluster_keys(self.ctx.clone(), &self.plan.catalog)
             .await?;
-        Ok(Box::pin(DataBlockStream::create(
-            self.plan.schema(),
-            None,
-            vec![],
-        )))
+
+        Ok(PipelineBuildResult::create())
     }
 }

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

@@ -256,7 +256,7 @@ impl CopyInterpreterV2 {
             files: vec![],
         };
 
-        let mut build_res = select_interpreter.create_new_pipeline().await?;
+        let mut build_res = select_interpreter.execute2().await?;
         let table = StageTable::try_create(stage_table_info)?;
 
         append2table(self.ctx.clone(), table.clone(), data_schema.clone(), &mut build_res, false, true)?;
@@ -272,7 +272,7 @@ 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.create_new_pipeline().await?;
+        let build_res = self.execute2().await?;
 
         let settings = self.ctx.get_settings();
         let query_need_abort = self.ctx.query_need_abort();
@@ -295,7 +295,7 @@ impl Interpreter for CopyInterpreterV2 {
         )))
     }
 
-    async fn create_new_pipeline(&self) -> Result<PipelineBuildResult> {
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
         match &self.plan {
             // TODO(xuanwo): extract them as a separate function.
             CopyPlanV2::IntoTable {

src/query/service/src/interpreters/interpreter_database_create.rs

@@ -23,6 +23,7 @@ use common_streams::DataBlockStream;
 use common_streams::SendableDataBlockStream;
 
 use crate::interpreters::Interpreter;
+use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 
@@ -45,7 +46,7 @@ impl Interpreter for CreateDatabaseInterpreter {
     }
 
     #[tracing::instrument(level = "debug", skip(self), fields(ctx.id = self.ctx.get_id().as_str()))]
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
         self.ctx
             .get_current_session()
             .validate_privilege(&GrantObject::Global, UserPrivilegeType::Create)
@@ -67,10 +68,6 @@ impl Interpreter for CreateDatabaseInterpreter {
         };
         catalog.create_database(self.plan.clone().into()).await?;
 
-        Ok(Box::pin(DataBlockStream::create(
-            self.plan.schema(),
-            None,
-            vec![],
-        )))
+        Ok(PipelineBuildResult::create())
     }
 }

src/query/service/src/interpreters/interpreter_database_drop.rs

@@ -22,6 +22,7 @@ use common_streams::DataBlockStream;
 use common_streams::SendableDataBlockStream;
 
 use crate::interpreters::Interpreter;
+use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 
@@ -42,7 +43,7 @@ impl Interpreter for DropDatabaseInterpreter {
         "DropDatabaseInterpreter"
     }
 
-    async fn execute(&self) -> Result<SendableDataBlockStream> {
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
         self.ctx
             .get_current_session()
             .validate_privilege(&GrantObject::Global, UserPrivilegeType::Drop)
@@ -51,10 +52,6 @@ impl Interpreter for DropDatabaseInterpreter {
         let catalog = self.ctx.get_catalog(&self.plan.catalog)?;
         catalog.drop_database(self.plan.clone().into()).await?;
 
-        Ok(Box::pin(DataBlockStream::create(
-            self.plan.schema(),
-            None,
-            vec![],
-        )))
+        Ok(PipelineBuildResult::create())
     }
 }