Merge pull request #7582 from zhang2014/refactor/interpreter_2

refactor(interpreter): remove sendable stream in interpreter

Author: mergify[bot]

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

@@ -18,6 +18,7 @@ pub mod deserializer;
 pub mod empty_source;
 pub mod file_splitter;
 pub mod multi_file_splitter;
+mod one_block_source;
 pub mod stream_source;
 pub mod stream_source_v2;
 pub mod sync_source;
@@ -33,6 +34,7 @@ pub use file_splitter::FileSplitter;
 pub use file_splitter::FileSplitterState;
 pub use multi_file_splitter::MultiFileSplitter;
 pub use multi_file_splitter::OperatorInfo;
+pub use one_block_source::OneBlockSource;
 pub use stream_source::StreamSource;
 pub use stream_source::StreamSourceNoSkipEmpty;
 pub use stream_source_v2::StreamSourceV2;

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

@@ -0,0 +1,66 @@
+// 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::sync::Arc;
+
+use common_datablocks::DataBlock;
+use common_exception::Result;
+use common_pipeline_core::processors::port::OutputPort;
+use common_pipeline_core::processors::processor::Event;
+use common_pipeline_core::processors::processor::ProcessorPtr;
+use common_pipeline_core::processors::Processor;
+
+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/benches/suites/mod.rs

@@ -37,8 +37,8 @@ pub async fn select_executor(sql: &str) -> Result<()> {
     let ctx = executor_session.create_query_context().await?;
 
     if let PlanNode::Select(plan) = PlanParser::parse(ctx.clone(), sql).await? {
-        let executor = SelectInterpreter::try_create(ctx, plan)?;
-        let mut stream = executor.execute().await?;
+        let executor = SelectInterpreter::try_create(ctx.clone(), plan)?;
+        let mut stream = executor.execute(ctx.clone()).await?;
         while let Some(_block) = stream.next().await {}
     } else {
         unreachable!()

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() {
@@ -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/interpreter.rs

@@ -12,14 +12,24 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
+use common_catalog::table_context::TableContext;
+use common_datavalues::DataSchema;
 use common_datavalues::DataSchemaRef;
 use common_datavalues::DataSchemaRefExt;
 use common_exception::ErrorCode;
 use common_exception::Result;
+use common_streams::DataBlockStream;
 use common_streams::SendableDataBlockStream;
 
+use crate::interpreters::ProcessorExecutorStream;
+use crate::pipelines::executor::ExecutorSettings;
+use crate::pipelines::executor::PipelineCompleteExecutor;
+use crate::pipelines::executor::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,18 +44,52 @@ 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>;
+    async fn execute(&self, ctx: Arc<QueryContext>) -> Result<SendableDataBlockStream> {
+        let mut build_res = self.execute2().await?;
 
-    /// 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()
-        )))
+        if build_res.main_pipeline.pipes.is_empty() {
+            return Ok(Box::pin(DataBlockStream::create(
+                self.schema(),
+                None,
+                vec![],
+            )));
+        }
+
+        let settings = ctx.get_settings();
+        let query_need_abort = ctx.query_need_abort();
+        let executor_settings = ExecutorSettings::try_create(&settings)?;
+        build_res.set_max_threads(settings.get_max_threads()? as usize);
+
+        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
+    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

@@ -18,13 +18,9 @@ use std::sync::RwLock;
 use common_datavalues::DataSchemaRef;
 use common_exception::Result;
 use common_legacy_planners::CallPlan;
-use common_pipeline_core::Pipeline;
-use common_streams::SendableDataBlockStream;
 
 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 +57,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 +68,14 @@ impl Interpreter for CallInterpreter {
             *schema = Some(last_schema);
         }
 
-        let mut pipeline = Pipeline::create();
-        func.eval(self.ctx.clone(), plan.args.clone(), &mut pipeline)
-            .await?;
+        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

@@ -18,10 +18,9 @@ use common_exception::Result;
 use common_legacy_planners::AlterTableClusterKeyPlan;
 use common_meta_types::GrantObject;
 use common_meta_types::UserPrivilegeType;
-use common_streams::DataBlockStream;
-use common_streams::SendableDataBlockStream;
 
 use super::Interpreter;
+use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 
@@ -42,7 +41,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 +67,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

@@ -18,10 +18,9 @@ use common_exception::Result;
 use common_legacy_planners::DropTableClusterKeyPlan;
 use common_meta_types::GrantObject;
 use common_meta_types::UserPrivilegeType;
-use common_streams::DataBlockStream;
-use common_streams::SendableDataBlockStream;
 
 use super::Interpreter;
+use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 
@@ -42,7 +41,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 +65,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())
     }
 }