refactor(interpreter): make lint

Author: zhang2014

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

@@ -18,11 +18,11 @@ 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;
 pub mod sync_source_receiver;
-mod one_block_source;
 
 pub use async_source::AsyncSource;
 pub use async_source::AsyncSourcer;
@@ -34,14 +34,14 @@ 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;
 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

@@ -13,20 +13,15 @@
 // 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::Event;
+use common_pipeline_core::processors::processor::ProcessorPtr;
 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>,

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/execute_query.rs

@@ -13,15 +13,20 @@
 // limitations under the License.
 
 use std::sync::Arc;
+
 use common_catalog::table_context::TableContext;
-use common_datavalues::{DataSchema, DataSchemaRef};
+use common_datavalues::DataSchema;
+use common_datavalues::DataSchemaRef;
 use common_datavalues::DataSchemaRefExt;
 use common_exception::ErrorCode;
 use common_exception::Result;
-use common_streams::{DataBlockStream, SendableDataBlockStream};
-use crate::interpreters::ProcessorExecutorStream;
-use crate::pipelines::executor::{ExecutorSettings, PipelineCompleteExecutor, PipelinePullingExecutor};
+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;

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

@@ -17,14 +17,9 @@ use std::sync::RwLock;
 
 use common_datavalues::DataSchemaRef;
 use common_exception::Result;
-use common_pipeline_core::Pipeline;
 use common_planners::CallPlan;
-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;
@@ -74,8 +69,12 @@ impl Interpreter for CallInterpreter {
         }
 
         let mut build_res = PipelineBuildResult::create();
-        func.eval(self.ctx.clone(), plan.args.clone(), &mut build_res.main_pipeline)
-            .await?;
+        func.eval(
+            self.ctx.clone(),
+            plan.args.clone(),
+            &mut build_res.main_pipeline,
+        )
+        .await?;
 
         Ok(build_res)
     }

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

@@ -18,11 +18,9 @@ use common_exception::Result;
 use common_meta_types::GrantObject;
 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::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 

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

@@ -18,11 +18,9 @@ use common_exception::Result;
 use common_meta_types::GrantObject;
 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::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 

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

@@ -86,7 +86,7 @@ pub fn append2table(
                     Err(cause) => Err(ErrorCode::PanicError(format!(
                         "Maybe panic while in commit insert. {}",
                         cause
-                    )))
+                    ))),
                 };
             }
 
@@ -103,7 +103,8 @@ pub fn execute_pipeline(ctx: Arc<QueryContext>, mut res: PipelineBuildResult) ->
     res.set_max_threads(ctx.get_settings().get_max_threads()? as usize);
     let mut pipelines = res.sources_pipelines;
     pipelines.push(res.main_pipeline);
-    let executor = PipelineCompleteExecutor::from_pipelines(query_need_abort, pipelines, executor_settings)?;
+    let executor =
+        PipelineCompleteExecutor::from_pipelines(query_need_abort, pipelines, executor_settings)?;
     executor.execute()
 }
 

src/query/service/src/interpreters/interpreter_common.rs

@@ -15,27 +15,22 @@ use std::collections::HashSet;
 use std::path::Path;
 use std::sync::Arc;
 
-use common_datablocks::DataBlock;
+use common_base::base::GlobalIORuntime;
+use common_base::base::TrySpawn;
 use common_datavalues::prelude::*;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_meta_types::UserStageInfo;
 use common_planners::ReadDataSourcePlan;
 use common_planners::SourceInfo;
 use common_planners::StageTableInfo;
-use common_streams::DataBlockStream;
-use common_streams::SendableDataBlockStream;
 use futures::TryStreamExt;
 use regex::Regex;
-use common_base::base::{GlobalIORuntime, TrySpawn};
 
 use super::append2table;
 use crate::interpreters::Interpreter;
-use crate::interpreters::interpreter_common::{execute_pipeline};
 use crate::interpreters::SelectInterpreterV2;
-use crate::pipelines::executor::ExecutorSettings;
-use crate::pipelines::executor::PipelineCompleteExecutor;
-use crate::pipelines::{Pipeline, PipelineBuildResult};
+use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 use crate::sql::plans::CopyPlanV2;
@@ -117,7 +112,11 @@ impl CopyInterpreterV2 {
         }
     }
 
-    async fn purge_files(ctx: Arc<QueryContext>, from: &ReadDataSourcePlan, files: &Vec<String>) -> Result<()> {
+    async fn purge_files(
+        ctx: Arc<QueryContext>,
+        from: &ReadDataSourcePlan,
+        files: &Vec<String>,
+    ) -> Result<()> {
         match &from.source_info {
             SourceInfo::StageSource(table_info) => {
                 if table_info.stage_info.copy_options.purge {
@@ -172,7 +171,11 @@ impl CopyInterpreterV2 {
         tracing::info!("copy_files_to_table from source: {:?}", read_source_plan);
 
         let from_table = self.ctx.build_table_from_source_plan(&read_source_plan)?;
-        from_table.read2(self.ctx.clone(), &read_source_plan, &mut build_res.main_pipeline)?;
+        from_table.read2(
+            self.ctx.clone(),
+            &read_source_plan,
+            &mut build_res.main_pipeline,
+        )?;
 
         let to_table = self.ctx.get_table(catalog_name, db_name, tbl_name).await?;
 
@@ -195,7 +198,9 @@ impl CopyInterpreterV2 {
                 let task = GlobalIORuntime::instance().spawn(async move {
                     // Commit
                     let operations = ctx.consume_precommit_blocks();
-                    to_table.commit_insertion(ctx.clone(), &catalog_name, operations, false).await?;
+                    to_table
+                        .commit_insertion(ctx.clone(), &catalog_name, operations, false)
+                        .await?;
 
                     // Purge
                     CopyInterpreterV2::purge_files(ctx, &from, &files).await
@@ -207,7 +212,7 @@ impl CopyInterpreterV2 {
                     Err(cause) => Err(ErrorCode::PanicError(format!(
                         "Maybe panic while in commit insert. {}",
                         cause
-                    )))
+                    ))),
                 };
             }
 
@@ -263,7 +268,14 @@ impl CopyInterpreterV2 {
         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)?;
+        append2table(
+            self.ctx.clone(),
+            table.clone(),
+            data_schema.clone(),
+            &mut build_res,
+            false,
+            true,
+        )?;
         Ok(build_res)
     }
 }
@@ -308,9 +320,18 @@ impl Interpreter for CopyInterpreterV2 {
                 }
 
                 tracing::info!("matched files: {:?}, pattern: {}", &files, pattern);
-                self.copy_files_to_table(catalog_name, database_name, table_name, from, files.clone()).await
+                self.copy_files_to_table(
+                    catalog_name,
+                    database_name,
+                    table_name,
+                    from,
+                    files.clone(),
+                )
+                .await
             }
-            CopyPlanV2::IntoStage { stage, from, path, .. } => self.execute_copy_into_stage(stage, path, from).await,
+            CopyPlanV2::IntoStage {
+                stage, from, path, ..
+            } => self.execute_copy_into_stage(stage, path, from).await,
         }
     }
 }