refactor(query): re-org query crates

Cargo.toml

@@ -46,6 +46,7 @@ members = [
     "src/query/legacy-expression",
     "src/query/legacy-planners",
     "src/query/settings",
+    "src/query/sql",
     "src/query/storages/cache",
     "src/query/storages/context",
     "src/query/storages/constants",
@@ -55,6 +56,7 @@ members = [
     "src/query/storages/hive-meta-store",
     "src/query/storages/index",
     "src/query/storages/preludes",
+    "src/query/storages/factory",
     "src/query/storages/share",
     "src/query/streams",
     "src/query/users",

src/query/catalog/Cargo.toml

@@ -27,3 +27,4 @@ common-storage = { path = "../../common/storage" }
 
 async-trait = "0.1.57"
 dyn-clone = "1.0.9"
+once_cell = "1.15.0"

src/query/catalog/src/catalog.rs

@@ -16,6 +16,7 @@ use std::any::Any;
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use common_base::base::Singleton;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_meta_app::schema::CountTablesReply;
@@ -49,6 +50,7 @@ use common_meta_app::schema::UpsertTableOptionReply;
 use common_meta_app::schema::UpsertTableOptionReq;
 use common_meta_types::MetaId;
 use dyn_clone::DynClone;
+use once_cell::sync::OnceCell;
 
 use crate::database::Database;
 use crate::table::Table;
@@ -57,16 +59,30 @@ use crate::table_function::TableFunction;
 
 pub const CATALOG_DEFAULT: &str = "default";
 
+static CATALOG_MANAGER: OnceCell<Singleton<Arc<CatalogManager>>> = OnceCell::new();
+
 pub struct CatalogManager {
     pub catalogs: HashMap<String, Arc<dyn Catalog>>,
 }
+
 impl CatalogManager {
     pub fn get_catalog(&self, catalog_name: &str) -> Result<Arc<dyn Catalog>> {
         self.catalogs
             .get(catalog_name)
             .cloned()
             .ok_or_else(|| ErrorCode::BadArguments(format!("not such catalog {}", catalog_name)))
     }
+
+    pub fn instance() -> Arc<CatalogManager> {
+        match CATALOG_MANAGER.get() {
+            None => panic!("CatalogManager is not init"),
+            Some(catalog_manager) => catalog_manager.get(),
+        }
+    }
+
+    pub fn set_instance(manager: Singleton<Arc<CatalogManager>>) {
+        CATALOG_MANAGER.set(manager).ok();
+    }
 }
 
 #[derive(Default, Clone)]

src/query/catalog/src/lib.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![deny(unused_crate_dependencies)]
-
 pub mod catalog;
 pub mod cluster_info;
 pub mod database;

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

@@ -17,6 +17,7 @@ pub mod transform_block_compact;
 pub mod transform_compact;
 pub mod transform_expression;
 pub mod transform_expression_executor;
+pub mod transform_limit;
 pub mod transform_sort_merge;
 pub mod transform_sort_partial;
 
@@ -25,5 +26,6 @@ pub use transform_block_compact::*;
 pub use transform_compact::*;
 pub use transform_expression::*;
 pub use transform_expression_executor::ExpressionExecutor;
+pub use transform_limit::*;
 pub use transform_sort_merge::*;
 pub use transform_sort_partial::*;

src/query/pipeline/transforms/src/processors/transforms/transform_expression.rs

@@ -27,7 +27,6 @@ use crate::processors::transforms::transform::Transform;
 use crate::processors::transforms::transform::Transformer;
 use crate::processors::transforms::ExpressionExecutor;
 
-pub type ProjectionTransform = ExpressionTransformImpl<true>;
 pub type ExpressionTransform = ExpressionTransformImpl<false>;
 
 pub struct ExpressionTransformImpl<const ALIAS_PROJECT: bool> {

src/query/pipeline/transforms/src/processors/transforms/transform_limit.rs

@@ -0,0 +1,206 @@
+// 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::ErrorCode;
+use common_exception::Result;
+use common_pipeline_core::processors::port::InputPort;
+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 TransformLimit;
+
+impl TransformLimit {
+    pub fn try_create(
+        limit: Option<usize>,
+        offset: usize,
+        input: Arc<InputPort>,
+        output: Arc<OutputPort>,
+    ) -> Result<ProcessorPtr> {
+        match (limit, offset) {
+            (None, 0) => Err(ErrorCode::LogicalError("It's a bug")),
+            (Some(_), 0) => OnlyLimitTransform::create(input, output, limit, offset),
+            (None, _) => OnlyOffsetTransform::create(input, output, limit, offset),
+            (Some(_), _) => OffsetAndLimitTransform::create(input, output, limit, offset),
+        }
+    }
+}
+
+const ONLY_LIMIT: usize = 0;
+const ONLY_OFFSET: usize = 1;
+const OFFSET_AND_LIMIT: usize = 2;
+
+type OnlyLimitTransform = TransformLimitImpl<ONLY_LIMIT>;
+type OnlyOffsetTransform = TransformLimitImpl<ONLY_OFFSET>;
+type OffsetAndLimitTransform = TransformLimitImpl<OFFSET_AND_LIMIT>;
+
+struct TransformLimitImpl<const MODE: usize> {
+    take_remaining: usize,
+    skip_remaining: usize,
+
+    input: Arc<InputPort>,
+    output: Arc<OutputPort>,
+
+    input_data_block: Option<DataBlock>,
+    output_data_block: Option<DataBlock>,
+}
+
+impl<const MODE: usize> TransformLimitImpl<MODE> {
+    pub fn create(
+        input: Arc<InputPort>,
+        output: Arc<OutputPort>,
+        limit: Option<usize>,
+        offset: usize,
+    ) -> Result<ProcessorPtr> {
+        Ok(ProcessorPtr::create(Box::new(Self {
+            input,
+            output,
+            input_data_block: None,
+            output_data_block: None,
+            skip_remaining: offset,
+            take_remaining: limit.unwrap_or(0),
+        })))
+    }
+
+    pub fn take_rows(&mut self, data_block: DataBlock) -> DataBlock {
+        let rows = data_block.num_rows();
+        debug_assert_ne!(self.take_remaining, 0);
+
+        if self.take_remaining >= rows {
+            self.take_remaining -= rows;
+            return data_block;
+        }
+
+        let remaining = self.take_remaining;
+        self.take_remaining = 0;
+        data_block.slice(0, remaining)
+    }
+
+    pub fn skip_rows(&mut self, data_block: DataBlock) -> Option<DataBlock> {
+        let rows = data_block.num_rows();
+
+        if self.skip_remaining >= rows {
+            self.skip_remaining -= rows;
+            return None;
+        }
+
+        match MODE {
+            OFFSET_AND_LIMIT => {
+                let offset = self.skip_remaining;
+                self.skip_remaining = 0;
+                let length = std::cmp::min(self.take_remaining, rows - offset);
+                self.take_remaining -= length;
+                Some(data_block.slice(offset, length))
+            }
+            _ => {
+                let offset = self.skip_remaining;
+                self.skip_remaining = 0;
+                Some(data_block.slice(offset, rows - offset))
+            }
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl<const MODE: usize> Processor for TransformLimitImpl<MODE> {
+    fn name(&self) -> String {
+        match MODE {
+            ONLY_LIMIT => "LimitTransform",
+            ONLY_OFFSET => "OffsetTransform",
+            OFFSET_AND_LIMIT => "OffsetAndLimitTransform",
+            _ => unreachable!(),
+        }
+        .to_string()
+    }
+
+    fn as_any(&mut self) -> &mut dyn Any {
+        self
+    }
+
+    fn event(&mut self) -> Result<Event> {
+        if self.output.is_finished() {
+            self.input.finish();
+            return Ok(Event::Finished);
+        }
+
+        if !self.output.can_push() {
+            self.input.set_not_need_data();
+            return Ok(Event::NeedConsume);
+        }
+
+        if let Some(data_block) = self.output_data_block.take() {
+            self.output.push_data(Ok(data_block));
+            return Ok(Event::NeedConsume);
+        }
+
+        if self.skip_remaining == 0 && self.take_remaining == 0 {
+            if MODE == ONLY_LIMIT || MODE == OFFSET_AND_LIMIT {
+                self.input.finish();
+                self.output.finish();
+                return Ok(Event::Finished);
+            }
+
+            if MODE == ONLY_OFFSET {
+                if self.input.is_finished() {
+                    self.output.finish();
+                    return Ok(Event::Finished);
+                }
+
+                if !self.input.has_data() {
+                    self.input.set_need_data();
+                    return Ok(Event::NeedData);
+                }
+
+                self.output.push_data(self.input.pull_data().unwrap());
+                return Ok(Event::NeedConsume);
+            }
+        }
+
+        if self.input_data_block.is_some() {
+            return Ok(Event::Sync);
+        }
+
+        if self.input.is_finished() {
+            self.output.finish();
+            return Ok(Event::Finished);
+        }
+
+        if !self.input.has_data() {
+            self.input.set_need_data();
+            return Ok(Event::NeedData);
+        }
+
+        self.input_data_block = Some(self.input.pull_data().unwrap()?);
+        Ok(Event::Sync)
+    }
+
+    fn process(&mut self) -> Result<()> {
+        if let Some(data_block) = self.input_data_block.take() {
+            self.output_data_block = match MODE {
+                ONLY_OFFSET => self.skip_rows(data_block),
+                ONLY_LIMIT => Some(self.take_rows(data_block)),
+                OFFSET_AND_LIMIT if self.skip_remaining != 0 => self.skip_rows(data_block),
+                OFFSET_AND_LIMIT => Some(self.take_rows(data_block)),
+                _ => unreachable!(),
+            }
+        }
+
+        Ok(())
+    }
+}

src/query/service/Cargo.toml

@@ -60,10 +60,12 @@ common-pipeline-sources = { path = "../pipeline/sources" }
 common-pipeline-transforms = { path = "../pipeline/transforms" }
 common-planner = { path = "../planner" }
 common-settings = { path = "../settings" }
+common-sql = { path = "../sql" }
 common-storage = { path = "../../common/storage" }
 common-storages-cache = { path = "../storages/cache" }
 common-storages-constants = { path = "../storages/constants" }
 common-storages-context = { path = "../storages/context" }
+common-storages-factory = { path = "../storages/factory" }
 common-storages-fuse = { path = "../storages/fuse" }
 common-storages-hive = { path = "../storages/hive", optional = true }
 common-storages-index = { path = "../storages/index" }

src/query/service/src/api/http/v1/logs.rs

@@ -26,8 +26,8 @@ use crate::sessions::QueryContext;
 use crate::sessions::SessionManager;
 use crate::sessions::SessionType;
 use crate::sessions::TableContext;
-use crate::storages::TableStreamReadWrap;
 use crate::storages::ToReadDataSourcePlan;
+use crate::utils::TableStreamReadWrap;
 
 // read log files from cfg.log.log_dir
 #[poem::handler]

src/query/service/src/api/http/v1/tenant_tables.rs

@@ -23,7 +23,6 @@ use poem::IntoResponse;
 use serde::Deserialize;
 use serde::Serialize;
 
-use crate::catalogs::CatalogManagerHelper;
 use crate::sessions::SessionManager;
 
 #[derive(Serialize, Deserialize, Eq, PartialEq, Debug, Default)]

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

@@ -53,9 +53,9 @@ use crate::interpreters::QueryFragmentsActions;
 use crate::pipelines::executor::ExecutorSettings;
 use crate::pipelines::executor::PipelineCompleteExecutor;
 use crate::pipelines::PipelineBuildResult;
+use crate::pipelines::PipelineBuilder as PipelineBuilderV2;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
-use crate::sql::executor::PipelineBuilder as PipelineBuilderV2;
 
 pub struct DataExchangeManager {
     config: Config,

src/query/service/src/api/rpc/flight_scatter_hash_v2.rs

@@ -24,9 +24,9 @@ use common_functions::scalars::FunctionContext;
 use common_functions::scalars::FunctionFactory;
 
 use crate::api::rpc::flight_scatter::FlightScatter;
-use crate::evaluator::EvalNode;
-use crate::evaluator::Evaluator;
-use crate::evaluator::TypedVector;
+use crate::sql::evaluator::EvalNode;
+use crate::sql::evaluator::Evaluator;
+use crate::sql::evaluator::TypedVector;
 use crate::sql::executor::PhysicalScalar;
 
 #[derive(Clone)]